"""Tests for Handlers.""" from contextlib import contextmanager from unittest.mock import call, patch import pytest from flask import g from freezegun import freeze_time from owsrequest.utils import mock_request from playlist import sound_recording_handlers from playlist.api import app from playlist.services.ows_permissions import ( 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_ANALYTICS_EXPECTED_RESPONSE, PLACEMENT_STREAMS_EXPECTED_RESPONSE, PLACEMENT_STREAMS_ON_DATE_EXPECTED_RESPONSE, TOP_PLACEMENTS_EXPECTED_RESPONSE, ) from tests.unit.fixtures.queries import ( PLACEMENT_ANALYTICS_DB_RESPONSE, PLACEMENT_STREAMS_DB_RESPONSE, PLACEMENT_STREAMS_ON_DATE_DB_RESPONSE, TOP_PLACEMENTS_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_sound_recording_top_placements(client): with mock_handler(TOP_PLACEMENTS_DB_RESPONSE): placements_response = client.get( "/sound-recording/QZK6M1903348/top-placements", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.json == TOP_PLACEMENTS_EXPECTED_RESPONSE assert placements_response.status_code == 200 def test_sound_recording_top_placements_stream_countries(client): with mock_handler(TOP_PLACEMENTS_DB_RESPONSE): with patch( "playlist.sound_recording_handlers.fetch_sound_recording_top_placements", return_value={"placements": TOP_PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "sound-recording/QZK6M1903348/top-placements?&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, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_sound_recording_top_placements_curator_countries(client): with mock_handler(TOP_PLACEMENTS_DB_RESPONSE): with patch( "playlist.sound_recording_handlers.fetch_sound_recording_top_placements", return_value={"placements": TOP_PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "sound-recording/QZK6M1903348/top-placements?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, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_sound_recording_top_placements_store_ids(client): with mock_handler(TOP_PLACEMENTS_DB_RESPONSE): with patch( "playlist.sound_recording_handlers.fetch_sound_recording_top_placements", return_value={"placements": TOP_PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "sound-recording/QZK6M1903348/top-placements?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, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_sound_recording_top_placements_playlist_types(client): with mock_handler(TOP_PLACEMENTS_DB_RESPONSE): with patch( "playlist.sound_recording_handlers.fetch_sound_recording_top_placements", return_value={"placements": TOP_PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "sound-recording/QZK6M1903348/top-placements?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, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_sound_recording_top_placements_playlist_appearances(client): with mock_handler(TOP_PLACEMENTS_DB_RESPONSE): with patch( "playlist.sound_recording_handlers.fetch_sound_recording_top_placements", return_value={"placements": TOP_PLACEMENTS_DB_RESPONSE, "total_count": 5}, ) as stub: placements_response = client.get( "sound-recording/QZK6M1903348/top-placements?playlist_appearances=current", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 assert stub.call_count == 1 assert stub.call_args[0] == ( { "isrc": "QZK6M1903348", "playlist_appearances": "current", "use_primary_playlist_type": False, "transfer_product_ownership_enabled": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_sound_recording_top_placements_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): placements_response = client.get( "sound-recording/QZK6M1903348/top-placements", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 401 def test_sound_recording_top_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( "sound-recording/QZK6M1903348/top-placements", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 403 def test_get_sound_recording_placement_analytics_bulk(client): with mock_handler(PLACEMENT_ANALYTICS_DB_RESPONSE): input_data = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": { "distributors": ["theorchard", "awal"], "stream_countries": ["US", "CA"], }, } response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLACEMENT_ANALYTICS_EXPECTED_RESPONSE assert response.status_code == 200 def test_get_sound_recording_placement_analytics_bulk_no_matching_rows(client): # When no placement rows match (e.g. a placement that doesn't appear under # the requested playlist_appearances filter) the endpoint must still return # an array, not null, so consumers can parse `data.placements`. with mock_handler([]): input_data = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": {"playlist_appearances": "current"}, } response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json == {"data": {"placements": []}} def test_get_sound_recording_placement_analytics_bulk_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): input_data = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": {}, } response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_sound_recording_placement_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 = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": {}, } placements_response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data ) assert placements_response.status_code == 403 def test_get_sound_recording_placement_analytics_bulk_with_sme_enabled(client): input_data = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": {}, } with ( patch( "playlist.sound_recording_handlers.is_show_sme_data_enabled" ) as mock_is_show_sme_data_enabled, patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_bulk_sound_recording_placement_analytics" ) as mock_fetch_analytics, ): mock_is_show_sme_data_enabled.return_value = True mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_analytics.return_value = PLACEMENT_ANALYTICS_DB_RESPONSE placements_response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 mock_is_show_sme_data_enabled.assert_called_once() mock_fetch_analytics.assert_called_once_with( { "sound_recording_playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "distributors": ["theorchard", "awal", "sme"], "storefront_defined": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_sound_recording_placement_analytics_bulk_with_sme_disabled(client): input_data = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": {}, } with ( patch( "playlist.sound_recording_handlers.is_show_sme_data_enabled" ) as mock_is_show_sme_data_enabled, patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_bulk_sound_recording_placement_analytics" ) as mock_fetch_analytics, ): mock_is_show_sme_data_enabled.return_value = False mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_analytics.return_value = PLACEMENT_ANALYTICS_DB_RESPONSE placements_response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 mock_is_show_sme_data_enabled.assert_called_once() mock_fetch_analytics.assert_called_once_with( { "sound_recording_playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "distributors": ["theorchard", "awal"], "storefront_defined": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ) def test_get_sound_recording_placement_analytics_bulk_with_storefront_defined(client): input_data = { "playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", }, { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", }, ], "params": {}, } with ( patch( "playlist.sound_recording_handlers.is_show_sme_data_enabled" ) as mock_is_show_sme_data_enabled, patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_bulk_sound_recording_placement_analytics" ) as mock_fetch_analytics, ): mock_is_show_sme_data_enabled.return_value = False mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_analytics.return_value = PLACEMENT_ANALYTICS_DB_RESPONSE placements_response = client.post( "/sound-recording/placement-analytics-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert placements_response.status_code == 200 mock_is_show_sme_data_enabled.assert_called_once() mock_fetch_analytics.assert_has_calls( [ call( { "sound_recording_playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "286", "store_playlist_id": "4Pej5WoIpT84lCqwgc5aUM", } ], "distributors": ["theorchard", "awal"], "storefront_defined": False, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), call( { "sound_recording_playlist_placements": [ { "isrc": "GBCTA1300103", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", } ], "distributors": ["theorchard", "awal"], "storefront_defined": True, }, { "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), ], True, ) # Tests for placement-streams-bulk endpoint def test_get_sound_recording_placement_streams_bulk(client): with mock_handler(PLACEMENT_STREAMS_DB_RESPONSE): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": {}, } response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLACEMENT_STREAMS_EXPECTED_RESPONSE assert response.status_code == 200 def test_get_sound_recording_placement_streams_bulk_empty_placements(client): input_data = { "playlist_placements": [], "params": {}, } with mock_handler(PLACEMENT_STREAMS_DB_RESPONSE): response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 400 assert response.json == {"error": "Missing playlist_placements"} def test_get_sound_recording_placement_streams_bulk_empty_permission_response(client): with patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources", return_value=None, ): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": {}, } response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 401 def test_get_sound_recording_placement_streams_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 = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": {}, } response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 403 def test_get_sound_recording_placement_streams_bulk_with_sme_enabled(client): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": {}, } with ( patch( "playlist.sound_recording_handlers.is_show_sme_data_enabled" ) as mock_is_show_sme_data_enabled, patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_playlist_placement_streams_aggregated" ) as mock_fetch_streams, ): mock_is_show_sme_data_enabled.return_value = True mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_streams.return_value = PLACEMENT_STREAMS_DB_RESPONSE response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 # Verify SME distributors are used assert mock_fetch_streams.call_args[0][0]["distributors"] == [ "theorchard", "awal", "sme", ] def test_get_sound_recording_placement_streams_bulk_with_stream_countries(client): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": { "stream_countries": ["US", "CA"], }, } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_playlist_placement_streams_aggregated" ) as mock_fetch_streams, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_streams.return_value = PLACEMENT_STREAMS_DB_RESPONSE response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 # Verify stream_countries is passed correctly assert mock_fetch_streams.call_args[0][0]["stream_countries"] == ["US", "CA"] def test_get_sound_recording_placement_streams_bulk_with_storefront(client): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", }, ], "params": {}, } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_playlist_placement_streams_aggregated" ) as mock_fetch_streams, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_streams.return_value = PLACEMENT_STREAMS_DB_RESPONSE response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 # Verify storefront_defined is True for placements with storefront assert mock_fetch_streams.call_args[0][0]["storefront_defined"] is True def test_get_sound_recording_placement_streams_bulk_with_and_without_storefront(client): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, { "isrc": "USRC12301932", "store_id": "1", "store_playlist_id": "pl.099d868325394f949586d8a071a084f1", "storefront": "US", }, ], "params": {}, } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_playlist_placement_streams_aggregated" ) as mock_fetch_streams, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_streams.return_value = PLACEMENT_STREAMS_DB_RESPONSE response = client.post( "/sound-recording/placement-streams-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 # Verify fetch_playlist_placement_streams_aggregated was called twice # (once for with storefront, once for without) assert mock_fetch_streams.call_count == 2 def test_get_sound_recording_placement_streams_on_date_bulk(client): with mock_handler(PLACEMENT_STREAMS_ON_DATE_DB_RESPONSE): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": {"date": "2026-05-01"}, } response = client.post( "/sound-recording/placement-streams-on-date-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.json == PLACEMENT_STREAMS_ON_DATE_EXPECTED_RESPONSE assert response.status_code == 200 def test_get_sound_recording_placement_streams_on_date_bulk_missing_date(client): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": {}, } response = client.post( "/sound-recording/placement-streams-on-date-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 400 assert response.json == {"error": "Missing date"} def test_get_sound_recording_placement_streams_on_date_bulk_with_stream_countries( client, ): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": { "date": "2026-05-01", "stream_countries": ["US", "CA"], }, } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_playlist_placement_streams_on_date" ) as mock_fetch_streams, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_streams.return_value = PLACEMENT_STREAMS_ON_DATE_DB_RESPONSE response = client.post( "/sound-recording/placement-streams-on-date-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert mock_fetch_streams.call_args[0][0]["stream_countries"] == ["US", "CA"] assert mock_fetch_streams.call_args[0][0]["date"] == "2026-05-01" def test_get_sound_recording_placement_streams_on_date_bulk_with_distributors( client, ): input_data = { "playlist_placements": [ { "isrc": "USRC12301932", "store_id": "286", "store_playlist_id": "37i9dQZF1DXcBWIGoYBM5M", }, ], "params": { "date": "2026-05-01", "distributors": ["theorchard", "awal"], }, } with ( patch( "playlist.services.ows_permissions.call_ows_permissions_get_profile_resources" ) as mock_get_permissions, patch( "playlist.sound_recording_handlers.fetch_playlist_placement_streams_on_date" ) as mock_fetch_streams, ): mock_get_permissions.return_value = [ { "id": "*", "name": "All Orchard Labels", "roles": ["Analytics"], "type": "Vendor", "uuid": "053a1a75-acc5-4cd8-9206-a194335d2afa", "vendorId": "*", } ] mock_fetch_streams.return_value = PLACEMENT_STREAMS_ON_DATE_DB_RESPONSE response = client.post( "/sound-recording/placement-streams-on-date-bulk", json=input_data, headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert mock_fetch_streams.call_args[0][0]["distributors"] == [ "theorchard", "awal", ]