"""Tests for Playlist Historical Tracklist endpoint.""" from contextlib import contextmanager from unittest.mock import patch import pytest from flask import g from owsrequest.utils import mock_request from playlist import 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.services.test_ows_permissions import OWS_PERMISSIONS_ALL_ACCESS_RESPONSE @contextmanager def mock_handler( snowflake_db_response={}, permissions_response=OWS_PERMISSIONS_ALL_ACCESS_RESPONSE, ): """Mock Snowflake query and permissions service.""" 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", ), ): 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 class TestPlaylistHistoricalTracklist: """Test suite for GET /playlist//placements/on-date endpoint.""" def test_get_historical_tracklist_success(self, client): """Test successfully getting historical tracklist for a playlist.""" historical_tracklist_response = [ { "isrc": "USXYZ1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 1, "previous_position": 5, "position_change": 4, # Improved from position 5 to 1 "peak_position": 1, "first_added_timestamp": "2025-10-15 00:00:00", "days_on_playlist_at_date": 17, "last_event_timestamp": "2025-11-01 00:00:00", "gsr_id": "test-gsr-id-1", "total_count": 50, }, { "isrc": "USABC9876543", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 2, "previous_position": 2, "position_change": 0, # No change "peak_position": 1, "first_added_timestamp": "2025-10-20 00:00:00", "days_on_playlist_at_date": 12, "last_event_timestamp": "2025-11-01 00:00:00", "gsr_id": "test-gsr-id-2", "total_count": 50, }, ] with mock_handler(historical_tracklist_response): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2025-11-01", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert "placements" in response.json["data"] assert "total_count" in response.json["data"] assert len(response.json["data"]["placements"]) == 2 assert response.json["data"]["total_count"] == 50 # Verify historical metrics are present assert response.json["data"]["placements"][0]["peak_position"] == 1 assert response.json["data"]["placements"][0]["position_change"] == 4 assert response.json["data"]["placements"][0]["days_on_playlist_at_date"] == 17 def test_get_historical_tracklist_spotify(self, client): """Test getting historical tracklist for Spotify playlist.""" historical_tracklist_response = [ { "isrc": "USXYZ1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-01 00:00:00", "days_on_playlist_at_date": 14, "last_event_timestamp": "2025-10-15 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with mock_handler(historical_tracklist_response): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2025-10-15", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"][0]["isrc"] == "USXYZ1234567" assert response.json["data"]["placements"][0]["current_position"] == 1 def test_get_historical_tracklist_apple_music(self, client): """Test getting historical tracklist for Apple Music playlist.""" historical_tracklist_response = [ { "isrc": "USABC9876543", "store_id": "1", "store_playlist_id": "pl.u-abc123xyz", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-10 00:00:00", "days_on_playlist_at_date": 10, "last_event_timestamp": "2025-10-20 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with mock_handler(historical_tracklist_response): response = client.get( "/playlist/pl.u-abc123xyz/placements/on-date?store_id=1&target_date=2025-10-20&storefront=us", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"][0]["isrc"] == "USABC9876543" assert response.json["data"]["placements"][0]["store_id"] == 1 def test_get_historical_tracklist_with_pagination(self, client): """Test historical tracklist with pagination parameters.""" historical_tracklist_response = [ { "isrc": "USXYZ1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 11, "previous_position": 15, "position_change": 4, "peak_position": 10, "first_added_timestamp": "2025-10-20 00:00:00", "days_on_playlist_at_date": 16, "last_event_timestamp": "2025-11-05 00:00:00", "gsr_id": "test-gsr-id", "total_count": 50, }, ] with mock_handler(historical_tracklist_response): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2025-11-05&limit=10&offset=10", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["total_count"] == 50 assert len(response.json["data"]["placements"]) == 1 def test_get_historical_tracklist_empty(self, client): """Test historical tracklist with no results.""" historical_tracklist_response = [] with mock_handler(historical_tracklist_response): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2020-01-01", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"] == [] assert response.json["data"]["total_count"] == 0 def test_get_historical_tracklist_old_date(self, client): """Test getting historical tracklist for a date far in the past.""" historical_tracklist_response = [ { "isrc": "USOLD1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 1, "previous_position": 3, "position_change": 2, "peak_position": 1, "first_added_timestamp": "2023-01-01 00:00:00", "days_on_playlist_at_date": 14, "last_event_timestamp": "2023-01-15 00:00:00", "gsr_id": "test-gsr-id", "total_count": 25, }, ] with mock_handler(historical_tracklist_response): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2023-01-15", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert response.json["data"]["placements"][0]["isrc"] == "USOLD1234567" assert response.json["data"]["total_count"] == 25 def test_get_historical_tracklist_with_stream_countries(self, client): """Test historical tracklist with stream_countries filter.""" historical_tracklist_response = [ { "isrc": "USXYZ1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-01 00:00:00", "days_on_playlist_at_date": 15, "last_event_timestamp": "2025-10-15 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with ( patch( "playlist.handlers.get_permissions", return_value={ "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), patch( "playlist.handlers.fetch_placements_by_store_playlist_id_on_date", return_value=(historical_tracklist_response, [], 1), ) as fetch_stub, ): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2025-10-15&streams_country=US&streams_country=MX", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert fetch_stub.call_count == 1 # Verify stream_countries was passed to the fetch function call_args = fetch_stub.call_args[0][0] assert "stream_countries" in call_args assert call_args["stream_countries"] == ["US", "MX"] assert call_args["store_id"] == "286" assert call_args["target_date"] == "2025-10-15" assert call_args["store_playlist_id"] == "37i9dQZF1DX4W3aJJYCDfV" def test_get_historical_tracklist_with_curator_countries(self, client): """Test historical tracklist with curator_countries filter.""" historical_tracklist_response = [ { "isrc": "USXYZ1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-01 00:00:00", "days_on_playlist_at_date": 15, "last_event_timestamp": "2025-10-15 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with ( patch( "playlist.handlers.get_permissions", return_value={ "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), patch( "playlist.handlers.fetch_placements_by_store_playlist_id_on_date", return_value=(historical_tracklist_response, [], 1), ) as fetch_stub, ): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2025-10-15&curator_country=DE&curator_country=FR", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert fetch_stub.call_count == 1 # Verify curator_countries was passed to the fetch function call_args = fetch_stub.call_args[0][0] assert "curator_countries" in call_args assert call_args["curator_countries"] == ["DE", "FR"] def test_get_historical_tracklist_with_both_stream_and_curator_countries( self, client ): """Test historical tracklist with both stream_countries and curator_countries filters.""" historical_tracklist_response = [ { "isrc": "USXYZ1234567", "store_id": "286", "store_playlist_id": "37i9dQZF1DX4W3aJJYCDfV", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-01 00:00:00", "days_on_playlist_at_date": 15, "last_event_timestamp": "2025-10-15 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with ( patch( "playlist.handlers.get_permissions", return_value={ "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), patch( "playlist.handlers.fetch_placements_by_store_playlist_id_on_date", return_value=(historical_tracklist_response, [], 1), ) as fetch_stub, ): response = client.get( "/playlist/37i9dQZF1DX4W3aJJYCDfV/placements/on-date?store_id=286&target_date=2025-10-15&streams_country=US&streams_country=CA&curator_country=US", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert fetch_stub.call_count == 1 # Verify both parameters were passed correctly call_args = fetch_stub.call_args[0][0] assert "stream_countries" in call_args assert call_args["stream_countries"] == ["US", "CA"] assert "curator_countries" in call_args assert call_args["curator_countries"] == ["US"] def test_get_historical_tracklist_apple_music_with_storefront(self, client): """Test Apple Music playlist with storefront - storefront takes precedence over stream_countries.""" historical_tracklist_response = [ { "isrc": "USABC9876543", "store_id": "1", "store_playlist_id": "pl.u-abc123xyz", "storefront": "us", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-10 00:00:00", "days_on_playlist_at_date": 10, "last_event_timestamp": "2025-10-20 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with ( patch( "playlist.handlers.get_permissions", return_value={ "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), patch( "playlist.handlers.fetch_placements_by_store_playlist_id_on_date", return_value=(historical_tracklist_response, [], 1), ) as fetch_stub, ): # Even though we pass stream_countries, storefront should take precedence response = client.get( "/playlist/pl.u-abc123xyz/placements/on-date?store_id=1&storefront=us&target_date=2025-10-20&streams_country=MX&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert fetch_stub.call_count == 1 # Verify Apple Music parameters call_args = fetch_stub.call_args[0][0] assert call_args["store_id"] == "1" assert call_args["storefront"] == "us" assert call_args["storefront_enabled"] is True # stream_countries should still be passed but will be ignored in SQL due to elif logic assert call_args.get("stream_countries") == ["MX"] def test_get_historical_tracklist_apple_music_without_storefront_enabled( self, client ): """Test Apple Music with storefront auto-enables storefront_enabled.""" historical_tracklist_response = [ { "isrc": "USABC9876543", "store_id": "1", "store_playlist_id": "pl.u-abc123xyz", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-10 00:00:00", "days_on_playlist_at_date": 10, "last_event_timestamp": "2025-10-20 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with ( patch( "playlist.handlers.get_permissions", return_value={ "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), patch( "playlist.handlers.fetch_placements_by_store_playlist_id_on_date", return_value=(historical_tracklist_response, [], 1), ) as fetch_stub, ): # Without storefront_enabled, stream_countries should work response = client.get( "/playlist/pl.u-abc123xyz/placements/on-date?store_id=1&storefront=us&target_date=2025-10-20&streams_country=US&streams_country=CA", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert fetch_stub.call_count == 1 call_args = fetch_stub.call_args[0][0] assert call_args["store_id"] == "1" assert call_args["storefront"] == "us" # storefront_enabled should be auto-set to True when storefront is provided for Apple Music assert call_args.get("storefront_enabled") is True assert call_args["stream_countries"] == ["US", "CA"] def test_get_historical_tracklist_apple_music_storefront_with_curator_countries( self, client ): """Test Apple Music with storefront_enabled - storefront takes precedence over curator_countries too.""" historical_tracklist_response = [ { "isrc": "USABC9876543", "store_id": "1", "store_playlist_id": "pl.u-abc123xyz", "storefront": "gb", "current_position": 1, "previous_position": None, "position_change": None, "peak_position": 1, "first_added_timestamp": "2025-10-10 00:00:00", "days_on_playlist_at_date": 10, "last_event_timestamp": "2025-10-20 00:00:00", "gsr_id": "test-gsr-id", "total_count": 1, }, ] with ( patch( "playlist.handlers.get_permissions", return_value={ "permission_label_ids": [], "permission_subaccount_ids": [], "permission_label_participant_ids": [], }, ), patch( "playlist.handlers.fetch_placements_by_store_playlist_id_on_date", return_value=(historical_tracklist_response, [], 1), ) as fetch_stub, ): response = client.get( "/playlist/pl.u-abc123xyz/placements/on-date?store_id=1&storefront=gb&target_date=2025-10-20&curator_country=US&curator_country=DE&storefront_enabled=true", headers=ALL_PERMISSIONS_EMPLOYEE_HEADERS, ) assert response.status_code == 200 assert fetch_stub.call_count == 1 call_args = fetch_stub.call_args[0][0] assert call_args["store_id"] == "1" assert call_args["storefront"] == "gb" assert call_args["storefront_enabled"] is True # curator_countries should still be passed but storefront takes precedence in SQL assert call_args.get("curator_countries") == ["US", "DE"]