"""Tests for Handlers.""" from contextlib import contextmanager from unittest.mock import MagicMock, call, patch import pytest from flask import g from freezegun import freeze_time from owsrequest.utils import mock_request from playlist import handlers from playlist.api import app from playlist.handlers import should_filter_by_top_10_apple_music_markets from playlist.services.ows_permissions import ( DEFAULT_PERMISSIONS, OWS_PERMISSIONS_PROFILE_URL, OWS_PERMISSIONS_SERVICE_NAME, ) from tests.integration.consts.permissions import ALL_PERMISSIONS_EMPLOYEE_HEADERS from tests.unit.fixtures.api_responses import ( PLACEMENT_BREAKDOWN_BY_COUNTRY_EXPECTED_RESPONSE, PLACEMENT_EXPECTED_RESPONSE, PLACEMENTS_BREAKDOWN_EXPECTED_RESPONSE, PLACEMENTS_COUNT_EXPECTED_RESPONSE, PLACEMENTS_EXPECTED_RESPONSE, PLACEMENTS_POSITION_TIME_SERIES_EXPECTED_RESPONSE, PLAYLIST_ANALYTICS_EXPECTED_RESPONSE, PLAYLIST_ANALYTICS_EXPECTED_RESPONSE_CORRECT_ORDER, PLAYLIST_ANALYTICS_TIMESERIES_EXPECTED_RESPONSE, PLAYLIST_ANALYTICS_TIMESERIES_EXPECTED_RESPONSE_CORRECT_ORDER, PLAYLIST_ANALYTICS_TIMESERIES_WITH_FOLLOWERS_EXPECTED_RESPONSE, PLAYLIST_METADATA_EXPECTED_RESPONSE, TOTAL_VS_PLACEMENT_STREAMS_EXPECTED_RESPONSE, TOTAL_VS_PLACEMENT_STREAMS_EXPECTED_RESPONSE_WITHOUT_AGG, ) from tests.unit.fixtures.queries import ( PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE, PLACEMENT_DB_RESPONSE, PLACEMENTS_BREAKDOWN_DB_RESPONSE, PLACEMENTS_COUNT_DB_RESPONSE, PLACEMENTS_DB_RESPONSE, PLACEMENTS_POSITION_TIME_SERIES_DB_RESPONSE, PLAYLIST_ANALYTICS_DB_RESPONSE, PLAYLIST_ANALYTICS_DB_RESPONSE_CORRECT_ORDER_WITH_STOREFRONT, PLAYLIST_ANALYTICS_DB_RESPONSE_CORRECT_ORDER_WITHOUT_STOREFRONT, PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE, PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE_CORRECT_ORDER_WITH_STOREFRONT, PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE_CORRECT_ORDER_WITHOUT_STOREFRONT, PLAYLIST_ANALYTICS_TIMESERIES_WITH_FOLLOWERS_DB_RESPONSE, PLAYLIST_METADATA_DB_RESPONSE, STREAMS_TIME_SERIES, TOTAL_VS_PLACEMENT_STREAMS_DB_RESPONSE, TOTAL_VS_PLACEMENT_STREAMS_WITHOUT_AGG_DB_RESPONSE, ) from tests.unit.services.test_ows_permissions import OWS_PERMISSIONS_ALL_ACCESS_RESPONSE @contextmanager def mock_handler( snowflake_db_response={}, permissions_response=OWS_PERMISSIONS_ALL_ACCESS_RESPONSE ): with ( patch( "playlist.connectors.snowflake.SnowflakeQuery.execute", return_value=snowflake_db_response, ) as snowflake, patch( "playlist.queries.fetch_queries.get_max_available_streaming_date", return_value="2021-06-01", ), patch( "playlist.features.is_insights_primary_playlist_type_enabled", return_value=False, ), ): mock_request.get( OWS_PERMISSIONS_SERVICE_NAME, OWS_PERMISSIONS_PROFILE_URL.format( profile_id=ALL_PERMISSIONS_EMPLOYEE_HEADERS["Orchard-Profile-Id"], profile_type=ALL_PERMISSIONS_EMPLOYEE_HEADERS["Orchard-Profile-Type"], ), status=200, response=permissions_response, ) yield snowflake @pytest.fixture def client(): """Return test client.""" test_client = app.test_client() class Ows: def __init__(self): self.correlation_id = "1" class RequestContext: def __init__(self): self.authorization = True self.context_type = "abcdef" with test_client.application.app_context(): g.ows = Ows() g.request_context = RequestContext() yield test_client def test_get_placements(client): with mock_handler(PLACEMENTS_DB_RESPONSE): placements_response = client.get( "/placements?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS ) assert placements_response.json == PLACEMENTS_EXPECTED_RESPONSE assert placements_response.status_code == 200 def test_get_placements_participant(client): with mock_handler(PLACEMENTS_DB_RESPONSE): placements_response = client.get( "/placements?global_participant_id=90097833-cc23-470e-8643-2268e156497f", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.json == PLACEMENTS_EXPECTED_RESPONSE assert placements_response.status_code == 200 def test_placements_stream_countries(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&streams_country=US&streams_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "stream_countries": ["US", "MX"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_stream_countries_participant(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?global_participant_id=90097833-cc23-470e-8643-2268e156497f&streams_country=US&streams_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "global_participant_id": "90097833-cc23-470e-8643-2268e156497f", "stream_countries": ["US", "MX"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_curator_countries(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&curator_country=US&curator_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "curator_countries": ["US", "MX"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_store_ids(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&store_id=1&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_ids": ["1", "286"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_playlist_types(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&playlist_type=USER_GENERATED&playlist_type=ALGORITHMIC", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "playlist_types": ["USER_GENERATED", "ALGORITHMIC"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_with_storefront_enabled(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&store_id=1&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_ids": ["1"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "storefront_enabled": True, "filter_by_top_10_apple_music_markets": True, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_with_storefront_disabled(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&store_id=1&storefront_enabled=false", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_ids": ["1"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "storefront_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_when_storefront_not_given(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&store_id=1&&store_id=2", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_ids": ["1", "2"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_when_no_store_ids_and_storefront_disabled(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&playlist_type=USER_GENERATED&storefront_enabled=false", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "playlist_types": ["USER_GENERATED"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "storefront_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_with_storefront_enabled_and_curator_country(client): """Ensure that filter_by_top_10_apple_music_markets is set to False when curator_country is passed in request""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&store_id=1&storefront_enabled=true&curator_country=US", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_ids": ["1"], "curator_countries": ["US"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "storefront_enabled": True, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_with_storefront_enabled_and_no_apple_music_id_in_store_ids(client): """Ensure that filter_by_top_10_apple_music_markets is set to False when no Apple Music id in store_ids""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements", return_value={"placements": PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "/placements?isrc=QZK6M1903348&store_id=286&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_ids": ["286"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, "insights_playlist_v2_current_past_enabled": False, "storefront_enabled": True, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_should_filter_by_top_10_apple_music_markets(): """Ensure that should_filter_by_top_10_apple_music_markets returns True""" query_params = { "storefront_enabled": True, "curator_countries": [], "store_ids": ["1"], } assert should_filter_by_top_10_apple_music_markets(query_params) def test_placements_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): placements_response = client.get( "/placements?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS ) assert placements_response.status_code == 401 def test_placements_forbidden_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): placements_response = client.get( "/placements?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS ) assert placements_response.status_code == 403 def test_get_placements_count(client): with mock_handler(PLACEMENTS_COUNT_DB_RESPONSE): placements_response = client.get( "/placements_count?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.json == PLACEMENTS_COUNT_EXPECTED_RESPONSE assert placements_response.status_code == 200 def test_placements_count_stream_countries(client): with mock_handler(PLACEMENTS_COUNT_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_count", return_value={"total_count": 5}, ) as stub: placements_response = client.get( "/placements_count?isrc=QZK6M1903348&streams_country=US&streams_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "stream_countries": ["US", "MX"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_count_curator_countries(client): with mock_handler(PLACEMENTS_COUNT_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_count", return_value={"total_count": 5}, ) as stub: placements_response = client.get( "/placements_count?isrc=QZK6M1903348&curator_country=US&curator_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "curator_countries": ["US", "MX"], "use_primary_playlist_type": False, "filter_incorrect_playlists": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_placements_count_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): placements_response = client.get( "/placements_count?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 401 def test_placements_count_forbidden_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): placements_response = client.get( "/placements_count?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 403 @freeze_time("2020-11-20") def test_get_placement_positions(client): with mock_handler(PLACEMENTS_POSITION_TIME_SERIES_DB_RESPONSE): response = client.get( "/playlist/74655ecbbb5583fd3d7898740d0620b69913c46d/placement/QM6N22019571/positions?start_date=2020-11-20&days=-20&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == { "data": {"positions": PLACEMENTS_POSITION_TIME_SERIES_EXPECTED_RESPONSE} } assert response.status_code == 200 def test_get_placement_streams(client): with mock_handler(STREAMS_TIME_SERIES): response = client.get( "/playlist/74655ecbbb5583fd3d7898740d0620b69913c46d/placement/QM6N22019571/streams?start_date=2020-01-23&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == {"data": {"streams": STREAMS_TIME_SERIES}} assert response.status_code == 200 def test_get_placement_streams_country(client): with patch( "playlist.handlers.get_permissions", return_value=DEFAULT_PERMISSIONS, ): with patch( "playlist.handlers.fetch_placement_streams_time_series", return_value=STREAMS_TIME_SERIES, ) as ts: response = client.get( "/playlist/74655ecbbb5583fd3d7898740d0620b69913c46d/placement/QM6N22019571/streams?start_date=2020-01-23&streams_country=DE&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == {"data": {"streams": STREAMS_TIME_SERIES}} assert response.status_code == 200 assert ts.call_count == 1 assert ts.call_args[0] == ( { "store_playlist_id": "74655ecbbb5583fd3d7898740d0620b69913c46d", "store_id": "286", "isrc": "QM6N22019571", "start_date": "2020-01-17", "end_date": "2020-01-23", "stream_countries": ["DE"], "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_placement_streams_countries_dates(client): with mock_handler(STREAMS_TIME_SERIES) as ts: with patch( "playlist.handlers.fetch_placement_streams_time_series", return_value=STREAMS_TIME_SERIES, ) as ts: response = client.get( "/playlist/74655ecbbb5583fd3d7898740d0620b69913c46d/placement/QM6N22019571/streams?start_date=2020-01-26&streams_country=DE&days=-4&streams_country=GB&streams_country=JP&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == {"data": {"streams": STREAMS_TIME_SERIES}} assert response.status_code == 200 assert ts.call_count == 1 assert ts.call_args[0] == ( { "store_playlist_id": "74655ecbbb5583fd3d7898740d0620b69913c46d", "store_id": "286", "isrc": "QM6N22019571", "start_date": "2020-01-23", "stream_countries": ["DE", "GB", "JP"], "end_date": "2020-01-26", "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_placement_streams_no_permissions(client): with mock_handler(None, None): response = client.get( "/playlist/74655ecbbb5583fd3d7898740d0620b69913c46d/placement/QM6N22019571/streams?start_date=2020-01-23&streams_country=DE&end_date=2020-01-26&streams_country=GB&streams_country=JP&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_placements_breakdown(client): with mock_handler(PLACEMENTS_BREAKDOWN_DB_RESPONSE): placements_breakdown_response = client.get( "/placements/breakdown?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_breakdown_response.json == PLACEMENTS_BREAKDOWN_EXPECTED_RESPONSE assert placements_breakdown_response.status_code == 200 def test_get_placements_breakdown_empty_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): response = client.get( "/placements/breakdown?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_placements_breakdown_forbidden_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): response = client.get( "/placements/breakdown?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 403 def test_get_total_vs_playlist_streams_by_store(client): with mock_handler(TOTAL_VS_PLACEMENT_STREAMS_DB_RESPONSE): total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert ( total_vs_playlist_streams_by_store_response.json == TOTAL_VS_PLACEMENT_STREAMS_EXPECTED_RESPONSE ) def test_get_total_vs_playlist_streams_global_participant(client): with mock_handler(TOTAL_VS_PLACEMENT_STREAMS_DB_RESPONSE): total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?global_participant_id=90097833-cc23-470e-8643-2268e156497f", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert ( total_vs_playlist_streams_by_store_response.json == TOTAL_VS_PLACEMENT_STREAMS_EXPECTED_RESPONSE ) def test_get_total_vs_playlist_streams_by_store_without_agg(client): with mock_handler(TOTAL_VS_PLACEMENT_STREAMS_WITHOUT_AGG_DB_RESPONSE): total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?isrc=QZK6M1903348", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert ( total_vs_playlist_streams_by_store_response.json == TOTAL_VS_PLACEMENT_STREAMS_EXPECTED_RESPONSE_WITHOUT_AGG ) def test_get_total_vs_playlist_streams_by_store_with_dates(client): with mock_handler(TOTAL_VS_PLACEMENT_STREAMS_DB_RESPONSE): with patch( "playlist.handlers.fetch_total_vs_playlist_streams_time_series", return_value=None, ) as ts: total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?isrc=QZK6M1903348&start_date=2020-01-01&days=32", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert total_vs_playlist_streams_by_store_response.status_code == 200 assert ts.call_count == 1 assert ts.call_args[0] == ( { "isrc": "QZK6M1903348", "start_date": "2020-01-01", "end_date": "2020-02-01", "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_total_vs_playlist_streams_by_store_with_countries(client): with mock_handler(TOTAL_VS_PLACEMENT_STREAMS_DB_RESPONSE): with patch( "playlist.handlers.fetch_total_vs_playlist_streams_time_series", return_value=None, ) as ts: total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?isrc=QZK6M1903348&streams_country=US", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert total_vs_playlist_streams_by_store_response.status_code == 200 assert ts.call_count == 1 assert ts.call_args[0] == ( { "start_date": "2021-05-26", "end_date": "2021-06-01", "isrc": "QZK6M1903348", "stream_countries": ["US"], "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_total_vs_playlist_streams_by_store_with_stores(client): with mock_handler(TOTAL_VS_PLACEMENT_STREAMS_DB_RESPONSE): with patch( "playlist.handlers.fetch_total_vs_playlist_streams_time_series", return_value=None, ) as ts: total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?isrc=QZK6M1903348&store_id=1", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert total_vs_playlist_streams_by_store_response.status_code == 200 assert ts.call_count == 1 assert ts.call_args[0] == ( { "start_date": "2021-05-26", "end_date": "2021-06-01", "isrc": "QZK6M1903348", "store_ids": ["1"], "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_total_vs_playlist_streams_by_store_empty_permission(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): total_vs_playlist_streams_by_store_response = client.get( "/placements/total_vs_playlist_streams_by_store?isrc=QZK6M1903348&store_id=1", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert total_vs_playlist_streams_by_store_response.status_code == 401 def test_get_placement_breakdown_by_country_empty_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): response = client.get( "/playlist/2/placement/3/breakdown-by-country?store_id=1", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_placement_breakdown_by_country_forbidden_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): response = client.get( "/playlist/2/placement/3/breakdown-by-country?store_id=1", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 403 def test_get_placement_breakdown_by_country(client): with mock_handler(PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE): response = client.get( "/playlist/pl.df3f10ca27b1479087de2cd3f9f6716b/placement/QMFME2004132/breakdown-by-country?store_id=1", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLACEMENT_BREAKDOWN_BY_COUNTRY_EXPECTED_RESPONSE def test_get_placement_breakdown_by_country_with_countries(client): with mock_handler(PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE): with patch( "playlist.handlers.fetch_placement_metrics_by_country", return_value={"storefronts": PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE}, ) as stub: response = client.get( ( "/playlist/pl.df3f10ca27b1479087de2cd3f9f6716b/placement/QMFME2004132/breakdown-by-country?" "store_id=1&streams_country=US&streams_country=MX" "&distributor=awal&distributor=theorchard&distributor=sme" ), headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QMFME2004132", "store_id": "1", "store_playlist_id": "pl.df3f10ca27b1479087de2cd3f9f6716b", "stream_countries": ["US", "MX"], "distributors": ["awal", "theorchard", "sme"], "insights_playlist_v2_current_past_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_placement_breakdown_by_country_with_current_placements(client): with mock_handler(PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE): with patch( "playlist.handlers.fetch_placement_metrics_by_country", return_value={"storefronts": PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE}, ) as stub: response = client.get( "/playlist/pl.df3f10ca27b1479087de2cd3f9f6716b/placement/QMFME2004132/breakdown-by-country?store_id=1&playlist_appearances=current", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QMFME2004132", "store_id": "1", "store_playlist_id": "pl.df3f10ca27b1479087de2cd3f9f6716b", "playlist_appearances": "current", "insights_playlist_v2_current_past_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_placement_breakdown_by_country_with_sorting(client): with mock_handler(PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE): with patch( "playlist.handlers.fetch_placement_metrics_by_country", return_value={"storefronts": PLACEMENT_BREAKDOWN_BY_COUNTRY_DB_RESPONSE}, ) as stub: response = client.get( "/playlist/pl.df3f10ca27b1479087de2cd3f9f6716b/placement/QMFME2004132/breakdown-by-country?store_id=1&sort_key=streams_last_7_days&sort_direction=DESC", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QMFME2004132", "store_id": "1", "store_playlist_id": "pl.df3f10ca27b1479087de2cd3f9f6716b", "sort_key": "streams_last_7_days", "sort_direction": "DESC", "insights_playlist_v2_current_past_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_placement(client): with mock_handler(PLACEMENT_DB_RESPONSE): placement_response = client.get( "/playlist/abc123/placement/QZK6M1903348?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.json == PLACEMENT_EXPECTED_RESPONSE assert placement_response.status_code == 200 def test_placement_stream_countries(client): with mock_handler(PLACEMENT_DB_RESPONSE): with patch( "playlist.handlers.fetch_placement", return_value={"placement": PLACEMENT_DB_RESPONSE}, ) as stub: placement_response = client.get( "/playlist/abc123/placement/QZK6M1903348?&streams_country=US&streams_country=MX&store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "store_playlist_id": "abc123", "store_id": "286", "stream_countries": ["US", "MX"], }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_placement_empty_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): response = client.get( "/playlist/abc123/placement/QZK6M1903348?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_recent_placements_empty_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): response = client.get( "/placements/recent", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_recent_placements_forbidden_permissions(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): response = client.get( "/placements/recent", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 403 def test_get_recent_placements(client): with mock_handler(PLACEMENTS_DB_RESPONSE): placement_response = client.get( "/placements/recent", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.json == PLACEMENTS_EXPECTED_RESPONSE assert placement_response.status_code == 200 def test_get_recent_placements_label_filter(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?label_id=26760", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "filter_label_ids": ["26760"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_subaccount_filter(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?subaccount_id=26760", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "filter_subaccount_ids": ["26760"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_store_filter(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_ids": ["286"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_playlisttype_filter(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?playlist_type=EDITORIAL", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "playlist_types": ["EDITORIAL"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_curator_country_comma_separated(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?curator_country=US,MX,GB", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "curator_countries": ["US", "MX", "GB"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_curator_country_repeated_keys(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?curator_country=US&curator_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "curator_countries": ["US", "MX"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_distributor_filter(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?distributor=theorchard", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "distributors": ["theorchard"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_storefront_enabled(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?store_id=1&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_ids": ["1"], "storefront_enabled": True, "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": True, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_sorting_parameters(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?sort_key=last_added_on_date&sort_direction=DESC", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "sort_key": "last_added_on_date", "sort_direction": "DESC", "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_curator_countries_filter(client): with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?sort_key=last_added_on_date&sort_direction=DESC&curator_country=DE&curator_country=IS", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "sort_key": "last_added_on_date", "sort_direction": "DESC", "curator_countries": ["DE", "IS"], "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_filtered_by_top_10_apple_music_markets(client): """ Ensure that filter_by_top_10_apple_music_markets is set to True when Apple Music is in store_ids and storefront_enabled is True """ with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?store_id=1&store_id=286&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_ids": ["1", "286"], "storefront_enabled": True, "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": True, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_recent_placements_not_filtered_by_top_10_apple_music_markets_when_curator_country_in_query( client, ): """Ensure that filter_by_top_10_apple_music_markets is set to False when curator_country is in query params""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_recent_placements", return_value=(PLACEMENTS_EXPECTED_RESPONSE, 0), ) as stub: placement_response = client.get( "/placements/recent?store_id=1&store_id=286&storefront_enabled=true&curator_country=US", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placement_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_ids": ["1", "286"], "curator_countries": ["US"], "storefront_enabled": True, "use_primary_playlist_type": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_playlist_metdata_bulk(client): with mock_handler(PLAYLIST_METADATA_DB_RESPONSE): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", }, { "store_id": "1", "store_playlist_id": "pl.53190961ecb54fb68ce05bff4b49ceca", }, ] } response = client.post( "/playlist-metadata-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLAYLIST_METADATA_EXPECTED_RESPONSE assert response.status_code == 200 def test_get_playlist_metadata_bulk_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } response = client.post( "/playlist-metadata-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_playlist_metadata_bulk_forbidden_permission_response( client, ): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } response = client.post("/playlist-metadata-bulk", json=input_data) assert response.status_code == 403 def test_get_playlist_metadata_bulk_with_storefront_defined(client): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_metadata" ) as mock_fetch_playlist_metadata, patch( "playlist.features.is_insights_primary_playlist_type_enabled" ) as mock_is_insights_primary_playlist_type_enabled, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_playlist_metadata.return_value = PLAYLIST_METADATA_DB_RESPONSE mock_is_insights_primary_playlist_type_enabled.return_value = False playlists_response = client.post( "/playlist-metadata-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert playlists_response.status_code == 200 mock_fetch_playlist_metadata.assert_has_calls( [ call( { "playlists": [ { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", } ], "storefront_defined": False, "use_primary_playlist_type": False, "insights_spotify_playlist_apollo_metadata_overrides_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), call( { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", } ], "storefront_defined": True, "use_primary_playlist_type": False, "insights_spotify_playlist_apollo_metadata_overrides_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), ], True, ) def test_get_playlist_metadata_bulk_uses_request_apollo_metadata_override_flag(client): input_data = { "playlists": [ { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "insights_spotify_playlist_apollo_metadata_overrides_enabled": True, } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_metadata" ) as mock_fetch_playlist_metadata, patch( "playlist.features.is_insights_primary_playlist_type_enabled" ) as mock_is_insights_primary_playlist_type_enabled, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_playlist_metadata.return_value = PLAYLIST_METADATA_DB_RESPONSE mock_is_insights_primary_playlist_type_enabled.return_value = False playlists_response = client.post( "/playlist-metadata-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert playlists_response.status_code == 200 mock_fetch_playlist_metadata.assert_called_once_with( { "playlists": [ { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", } ], "storefront_defined": False, "use_primary_playlist_type": False, "insights_spotify_playlist_apollo_metadata_overrides_enabled": True, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_playlist_analytics_bulk(client): with mock_handler(PLAYLIST_ANALYTICS_DB_RESPONSE): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", }, { "store_id": "1", "store_playlist_id": "pl.53190961ecb54fb68ce05bff4b49ceca", }, ] } response = client.post( "/playlist/analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLAYLIST_ANALYTICS_EXPECTED_RESPONSE assert response.status_code == 200 def test_get_playlist_analytics_bulk_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } response = client.post( "/playlist/analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_playlist_analytics_bulk_forbidden_permission_response( client, ): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } response = client.post("/playlist/analytics-bulk", json=input_data) assert response.status_code == 403 def test_get_playlist_analytics_bulk_and_order_kept(client): with ( patch( "playlist.handlers.fetch_bulk_playlist_analytics" ) as mock_fetch_playlist_analytics, mock_handler(), ): input_data = { "playlists": [ { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", "store_id": "1", }, { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5ac", "store_id": "1", "storefront": "US", }, { "store_playlist_id": "pl.53190961ecb54fb68ce05bff4b49ceca", "store_id": "2", }, { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", "store_id": "1", "storefront": "AD", }, ] } mock_fetch_playlist_analytics.side_effect = [ PLAYLIST_ANALYTICS_DB_RESPONSE_CORRECT_ORDER_WITHOUT_STOREFRONT, PLAYLIST_ANALYTICS_DB_RESPONSE_CORRECT_ORDER_WITH_STOREFRONT, ] response = client.post( "/playlist/analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLAYLIST_ANALYTICS_EXPECTED_RESPONSE_CORRECT_ORDER assert response.status_code == 200 def test_get_playlist_analytics_bulk_with_storefront_defined(client): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", "stream_countries": ["UK"], }, ] } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_analytics" ) as mock_fetch_playlist_analytics, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_playlist_analytics.return_value = PLAYLIST_ANALYTICS_DB_RESPONSE playlists_response = client.post( "/playlist/analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert playlists_response.status_code == 200 mock_fetch_playlist_analytics.assert_has_calls( [ call( { "playlists": [ { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", "row_number": 1, } ], "storefront_defined": False, "stream_countries": None, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), call( { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", "row_number": 0, } ], "storefront_defined": True, "stream_countries": None, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), call( { "playlists": [ { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", "row_number": 2, } ], "storefront_defined": False, "stream_countries": ["UK"], "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), ], True, ) def test_get_playlist_analytics_bulk_timeseries(client): with mock_handler(PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", }, { "store_id": "1", "store_playlist_id": "pl.53190961ecb54fb68ce05bff4b49ceca", }, ] } response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLAYLIST_ANALYTICS_TIMESERIES_EXPECTED_RESPONSE assert response.status_code == 200 def test_get_playlist_analytics_bulk_timeseries_with_followers(client): with mock_handler(PLAYLIST_ANALYTICS_TIMESERIES_WITH_FOLLOWERS_DB_RESPONSE): input_data = { "playlists": [ { "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", } ], "start_date": "2023-10-01", "end_date": "2023-10-31", } response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 playlist = response.json["data"]["playlists"][0] assert "followers_total_array" in playlist assert "followers_daily_change_array" in playlist assert len(playlist["followers_total_array"]) == 2 assert playlist["followers_total_array"][1]["followers"] == 10000 assert playlist["followers_daily_change_array"][1]["daily_change"] == 200 assert ( response.json == PLAYLIST_ANALYTICS_TIMESERIES_WITH_FOLLOWERS_EXPECTED_RESPONSE ) def test_get_playlist_analytics_timeseries_bulk_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_playlist_analytics_timeseries_bulk_forbidden_permission_response( client, ): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=[ { "id": "*", "name": "All Orchard Labels", "roles": ["NotAnalytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ], ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ] } response = client.post("/playlist/analytics-bulk-timeseries", json=input_data) assert response.status_code == 403 def test_get_playlist_analytics_timeseries_bulk_and_order_kept(client): with ( patch( "playlist.handlers.fetch_bulk_playlist_analytics_timeseries" ) as mock_fetch_bulk_playlist_analytics_timeseries, mock_handler(), ): input_data = { "playlists": [ { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", "store_id": "1", }, { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5ac", "store_id": "1", "storefront": "US", }, { "store_playlist_id": "pl.53190961ecb54fb68ce05bff4b49ceca", "store_id": "2", }, { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", "store_id": "1", "storefront": "AD", }, ] } mock_fetch_bulk_playlist_analytics_timeseries.side_effect = [ PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE_CORRECT_ORDER_WITHOUT_STOREFRONT, PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE_CORRECT_ORDER_WITH_STOREFRONT, ] response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert ( response.json == PLAYLIST_ANALYTICS_TIMESERIES_EXPECTED_RESPONSE_CORRECT_ORDER ) assert response.status_code == 200 def test_get_playlist_analytics_timeseries_bulk_with_storefront_defined(client): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, { "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "stream_countries": ["UK"], } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_analytics_timeseries" ) as mock_fetch_bulk_playlist_analytics_timeseries, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_bulk_playlist_analytics_timeseries.return_value = ( PLAYLIST_ANALYTICS_TIMESERIES_DB_RESPONSE ) playlists_response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert playlists_response.status_code == 200 mock_fetch_bulk_playlist_analytics_timeseries.assert_has_calls( [ call( { "playlists": [ { "row_number": 1, "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, { "row_number": 2, "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "start_date": None, "end_date": None, "storefront_defined": False, "stream_countries": ["UK"], "by_dimension": None, "order_by": "activity_date", "order_dir": "ASC", }, { "permission_label_ids": [], "permission_label_participant_ids": [], "permission_subaccount_ids": [], }, ), call( { "playlists": [ { "row_number": 0, "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", } ], "end_date": None, "start_date": None, "storefront_defined": True, "stream_countries": ["UK"], "by_dimension": None, "order_by": "activity_date", "order_dir": "ASC", }, { "permission_label_ids": [], "permission_label_participant_ids": [], "permission_subaccount_ids": [], }, ), ], True, ) def test_get_playlist_analytics_timeseries_bulk_returns_empty_when_fetch_returns_none( client, ): """When fetch_bulk_playlist_analytics_timeseries returns None, the endpoint should return 200 with an empty playlists list instead of raising a TypeError.""" with ( patch( "playlist.handlers.fetch_bulk_playlist_analytics_timeseries", return_value=None, ), mock_handler(), ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", } ] } response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["playlists"] == [] def test_get_playlist_analytics_timeseries_bulk_by_market_returns_empty_when_fetch_returns_none( client, ): """When fetch_bulk_playlist_analytics_timeseries returns None with by_dimension=MARKET, the endpoint should return 200 with an empty playlists list instead of raising a TypeError. """ with ( patch( "playlist.handlers.fetch_bulk_playlist_analytics_timeseries", return_value=None, ), mock_handler(), ): input_data = { "playlists": [ { "store_id": "1", "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", } ], "by_dimension": "MARKET", } response = client.post( "/playlist/analytics-bulk-timeseries", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["playlists"] == [] def test_get_placements_by_store_playlist_id_basic(client): """Test GET /playlist//placements with basic params.""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), # Added empty placeholder array ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/placements?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "use_primary_playlist_type": False, "insights_playlist_v2_current_past_enabled": False, "insights_playlist_page_hourly_playlists_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=True, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=True, ) def test_get_placements_by_store_playlist_id_with_storefront( mock_hourly_flag, mock_apple_music_flag, client ): """Test GET /playlist//placements with Apple Music storefront.""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), # Added empty placeholder array ) as stub: response = client.get( "/playlist/pl.u-abc123/placements?store_id=1&storefront=us&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_playlist_id": "pl.u-abc123", "store_id": "1", "storefront": "us", "curator_countries": ["us"], "use_primary_playlist_type": False, "insights_playlist_v2_current_past_enabled": False, "insights_playlist_page_hourly_playlists_enabled": True, "storefront_enabled": True, "filter_by_top_10_apple_music_markets": False, # False because store_id is singular not store_ids list }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=False, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=True, ) def test_get_placements_by_store_playlist_id_apple_music_blocked_when_apple_music_flag_off( mock_hourly_flag, mock_apple_music_flag, client ): """Test empty results when accessing Apple Music playlist with apple_music flag disabled.""" with mock_handler(): response = client.get( "/playlist/pl.u-abc123/placements?store_id=1&storefront=us", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"] == [] assert response.json["data"]["placeholder_placements"] == [] assert response.json["data"]["total_count"] == 0 @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=True, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=False, ) def test_get_placements_by_store_playlist_id_apple_music_blocked_when_hourly_flag_off( mock_hourly_flag, mock_apple_music_flag, client ): """Test empty results when accessing Apple Music playlist with hourly flag disabled.""" with mock_handler(): response = client.get( "/playlist/pl.u-abc123/placements?store_id=1&storefront=us", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"] == [] assert response.json["data"]["placeholder_placements"] == [] assert response.json["data"]["total_count"] == 0 @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=False, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=False, ) def test_get_placements_by_store_playlist_id_apple_music_blocked_when_both_flags_off( mock_hourly_flag, mock_apple_music_flag, client ): """Test empty results when accessing Apple Music playlist with both flags disabled.""" with mock_handler(): response = client.get( "/playlist/pl.u-abc123/placements?store_id=1&storefront=us", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"] == [] assert response.json["data"]["placeholder_placements"] == [] assert response.json["data"]["total_count"] == 0 @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=False, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=False, ) def test_get_placements_by_store_playlist_id_spotify_allowed_regardless_of_flags( mock_hourly_flag, mock_apple_music_flag, client ): """Test Spotify playlists work regardless of Apple Music flags.""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), ) as stub: response = client.get( "/playlist/37i9dQZEVXbMDoHDwVN2tF/placements?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=True, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=True, ) def test_get_placements_by_store_playlist_id_apple_music_auto_curator_countries_from_stream( mock_hourly_flag, mock_apple_music_flag, client ): """Test curator_countries auto-populated from stream_countries for Apple Music.""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), ) as stub: response = client.get( "/playlist/pl.u-abc123/placements?store_id=1&streams_country=us&streams_country=gb", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 # Verify curator_countries was auto-populated from stream_countries assert stub.call_args[0][0]["curator_countries"] == ["us", "gb"] assert stub.call_args[0][0]["stream_countries"] == ["us", "gb"] def test_get_placements_by_store_playlist_id_with_filters(client): """Test GET /playlist//placements with multiple filters.""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), # Added empty placeholder array ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/placements?" "store_id=286&playlist_type=EDITORIAL&curator_country=US" "&sort_key=streams_last_7_days&sort_direction=DESC&limit=50&offset=10", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "playlist_types": ["EDITORIAL"], "curator_countries": ["US"], "sort_key": "streams_last_7_days", "sort_direction": "DESC", "limit": "50", "offset": "10", "use_primary_playlist_type": False, "insights_playlist_v2_current_past_enabled": False, "insights_playlist_page_hourly_playlists_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_post_placements_by_store_playlist_id_with_json_filters(client): """Test POST /playlist//placements with JSON body filters.""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), ) as stub: response = client.post( "/playlist/37i9dQZF1DX2apWzyECwyZ/placements?store_id=286", json={ "playlist_type": ["EDITORIAL", "ALGORITHMIC"], "curator_country": ["US", "GB"], "distributor": ["theorchard", "awal"], }, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "playlist_types": ["EDITORIAL", "ALGORITHMIC"], "curator_countries": ["US", "GB"], "distributors": ["theorchard", "awal"], "use_primary_playlist_type": False, "insights_playlist_v2_current_past_enabled": False, "insights_playlist_page_hourly_playlists_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_post_placements_by_store_playlist_id_with_query_params(client): """Test POST /playlist//placements with query params (no JSON body filters).""" with mock_handler(PLACEMENTS_DB_RESPONSE): with patch( "playlist.handlers.fetch_placements_by_store_playlist_id", return_value=( PLACEMENTS_EXPECTED_RESPONSE, [], 5, ), ) as stub: response = client.post( "/playlist/37i9dQZF1DX2apWzyECwyZ/placements?" "store_id=286&playlist_type=EDITORIAL&distributor=theorchard&distributor=awal", json={}, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 # Should parse multi-valued params from query string when JSON body is empty assert stub.call_args[0] == ( { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "playlist_types": ["EDITORIAL"], "distributors": ["theorchard", "awal"], "use_primary_playlist_type": False, "insights_playlist_v2_current_past_enabled": False, "insights_playlist_page_hourly_playlists_enabled": False, "filter_by_top_10_apple_music_markets": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) @patch( "playlist.handlers.is_insights_playlist_page_apple_music_playlists_enabled", return_value=True, ) @patch( "playlist.queries.fetch_queries.is_insights_playlist_page_non_priority_playlists_enabled", return_value=True, ) @patch( "playlist.handlers.is_insights_playlist_page_hourly_playlists_enabled", return_value=True, ) def test_get_playlist_ids( mock_hourly_flag, mock_non_priority_flag, mock_apple_music_flag, client ): """Test the /playlist-ids endpoint returns all playlist IDs.""" mock_cursor = MagicMock() mock_cursor.fetchall.return_value = [ { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", "store_id": "1", }, {"store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", "store_id": "286"}, {"store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286"}, ] with mock_handler(mock_cursor): response = client.get("/playlist-ids") expected_response = { "data": { "playlists": [ { "store_playlist_id": "pl.f4d106fed2bd41149aaacabb233eb5eb", "store_id": 1, }, {"store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", "store_id": 286}, {"store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": 286}, ] } } assert response.json == expected_response assert response.status_code == 200 def test_company_brand_for_playlist_placement_for_default(client): """Test basic company brand fetch for playlist placements.""" with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_placements_company_brands" ) as mock_fetch, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch.return_value = [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": 286, "isrc": "USUM72005928", "row_number": 0, "brand_ids_array": ["9b36a627-4be9-4762-b790-5628676f5f90"], } ] input_data = { "playlist_placements": [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "isrc": "USUM72005928", }, ] } response = client.post( "/playlist/placement/company-brand", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 response_body = response.json["data"]["playlist_placements"] assert len(response_body) == 1 assert "row_number" not in response_body[0] def test_company_brand_for_playlist_placement_for_missing_placements(client): """Test that endpoint returns 400 when playlist_placements is empty.""" input_data = { "playlist_placements": [], } response = client.post( "/playlist/placement/company-brand", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 400 response_body = response.json assert response_body == {"error": "Missing placements"} def test_company_brand_for_playlist_placement_with_storefront(client): """Test company brand fetch with storefront parameter.""" with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_placements_company_brands" ) as mock_fetch, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch.return_value = [ { "store_playlist_id": "pl.56b8dbaf59a9471bbf0af144dc7a0f2b", "store_id": 1, "isrc": "USUM72005928", "row_number": 0, "brand_ids_array": ["ae886607-ae74-42f4-b7a7-d33f6a6288f4"], "storefront": "US", } ] input_data = { "playlist_placements": [ { "store_playlist_id": "pl.56b8dbaf59a9471bbf0af144dc7a0f2b", "store_id": "1", "storefront": "US", "isrc": "USUM72005928", }, ] } response = client.post( "/playlist/placement/company-brand", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 response_body = response.json["data"]["playlist_placements"] assert len(response_body) == 1 assert "row_number" not in response_body[0] def test_company_brand_for_playlist_placement_with_and_without_storefront(client): """Test company brand fetch with mixed storefront and non-storefront placements.""" with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_placements_company_brands" ) as mock_fetch, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] # Mock both calls for storefront and non-storefront separations mock_fetch.side_effect = [ [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": 286, "isrc": "USUM72005928", "row_number": 0, "brand_ids_array": [], } ], [ { "store_playlist_id": "pl.56b8dbaf59a9471bbf0af144dc7a0f2b", "store_id": 1, "isrc": "USUM72005928", "row_number": 1, "brand_ids_array": ["9b36a627-4be9-4762-b790-5628676f5f90"], "storefront": "US", } ], ] input_data = { "playlist_placements": [ { "store_playlist_id": "37i9dQZF1DX2apWzyECwyZ", "store_id": "286", "isrc": "USUM72005928", }, { "store_playlist_id": "pl.56b8dbaf59a9471bbf0af144dc7a0f2b", "store_id": "1", "storefront": "US", "isrc": "USUM72005928", }, ] } response = client.post( "/playlist/placement/company-brand", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 response_body = response.json["data"]["playlist_placements"] assert len(response_body) == 2 assert "row_number" not in response_body[0] assert "row_number" not in response_body[1] def test_company_brand_for_playlist_placement_order_is_preserved(client): """Test that output order matches input order (via row_number tracking).""" with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.handlers.fetch_bulk_playlist_placements_company_brands" ) as mock_fetch, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] # Return results in a different order to test sorting mock_fetch.side_effect = [ [ { "store_playlist_id": "pl.2", "store_id": 1, "isrc": "ISRC2", "row_number": 1, "brand_ids_array": ["9b36a627-4be9-4762-b790-5628676f5f90"], "storefront": "US", } ], [ { "store_playlist_id": "pl.1", "store_id": 286, "isrc": "ISRC1", "row_number": 0, "brand_ids_array": [], } ], ] input_data = { "playlist_placements": [ { "store_playlist_id": "pl.1", "store_id": "286", "isrc": "ISRC1", }, { "store_playlist_id": "pl.2", "store_id": "1", "storefront": "US", "isrc": "ISRC2", }, ] } response = client.post( "/playlist/placement/company-brand", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 response_body = response.json["data"]["playlist_placements"] assert len(response_body) == 2 # Verify order matches input (row_number 0, then 1) assert response_body[0]["isrc"] == "ISRC1" assert response_body[1]["isrc"] == "ISRC2" # Verify row_number was removed from output assert "row_number" not in response_body[0] assert "row_number" not in response_body[1] # --- Playlist Demographics Tests --- def test_get_playlist_demographics(client): """Test basic playlist demographics endpoint""" mock_demographics_response = [ { "total_streams": 1000000, "percent_male": 55, "percent_female": 45, "percent_under_18": 10, "percent_18_22": 25, "percent_23_27": 30, "percent_28_34": 20, "percent_35_44": 10, "percent_45_59": 4, "percent_60_plus": 1, "percent_age_unknown": 0, } ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics", return_value=mock_demographics_response, ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 response_body = response.json["data"]["demographics"] assert len(response_body) == 1 assert response_body[0]["total_streams"] == 1000000 assert response_body[0]["percent_male"] == 55 def test_get_playlist_demographics_by_country(client): """Test playlist demographics endpoint with by_country=true""" mock_demographics_response = [ { "country_code": "US", "total_streams": 500000, "percent_male": 52, "percent_female": 48, }, { "country_code": "GB", "total_streams": 200000, "percent_male": 58, "percent_female": 42, }, ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics_by_country", return_value=mock_demographics_response, ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/?store_id=286&by_country=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 response_body = response.json["data"]["demographics"] assert len(response_body) == 2 assert response_body[0]["country_code"] == "US" assert response_body[1]["country_code"] == "GB" def test_get_playlist_demographics_missing_store_id(client): """Test that store_id is required""" with mock_handler({}): response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 400 assert "store_id is required" in response.json["error"] def test_get_playlist_demographics_with_date_range(client): """Test playlist demographics with start_date and end_date""" mock_demographics_response = [ { "total_streams": 100000, "percent_male": 50, "percent_female": 50, } ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics", return_value=mock_demographics_response, ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/?store_id=286&start_date=2025-01-01&end_date=2025-01-31", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 # Check that the date params were passed through call_params = stub.call_args[0][0] assert call_params["start_date"] == "2025-01-01" assert call_params["end_date"] == "2025-01-31" def test_get_playlist_demographics_default_date_range(client): """Test that default date range is applied when no dates provided""" mock_demographics_response = [ { "total_streams": 100000, "percent_male": 50, "percent_female": 50, } ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics", return_value=mock_demographics_response, ) as stub: with patch( "playlist.queries.fetch_queries.get_max_available_streaming_date", return_value="2025-12-15", ): response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 # Check that default 28-day range was applied call_params = stub.call_args[0][0] assert call_params["end_date"] == "2025-12-15" assert call_params["start_date"] == "2025-11-18" # 27 days before end_date def test_get_playlist_demographics_with_storefront(client): """Test playlist demographics with storefront parameter""" mock_demographics_response = [ { "total_streams": 50000, "percent_male": 60, "percent_female": 40, } ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics", return_value=mock_demographics_response, ) as stub: response = client.get( "/playlist/pl.abc123/demographics/?store_id=1&storefront=US", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 call_params = stub.call_args[0][0] assert call_params["storefront"] == "US" assert call_params["store_id"] == "1" def test_get_playlist_demographics_with_stream_countries(client): """Test playlist demographics with streams_countries parameter""" mock_demographics_response = [ { "total_streams": 75000, "percent_male": 55, "percent_female": 45, } ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics", return_value=mock_demographics_response, ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/?store_id=286&stream_countries=US,GB,AU", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 call_params = stub.call_args[0][0] assert call_params["stream_countries"] == ["US", "GB", "AU"] def test_get_playlist_demographics_with_empty_stream_countries(client): """Test playlist demographics with no streams_countries parameter results in empty list""" mock_demographics_response = [ { "total_streams": 100000, "percent_male": 50, "percent_female": 50, } ] with mock_handler({}): with patch( "playlist.handlers.fetch_playlist_demographics", return_value=mock_demographics_response, ) as stub: response = client.get( "/playlist/37i9dQZF1DX2apWzyECwyZ/demographics/?store_id=286", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert stub.call_count == 1 call_params = stub.call_args[0][0] assert call_params["stream_countries"] == []