from typing import Any import pytest from pydantic import SecretStr from pytest_mock import MockerFixture from resonance_engine.adapters.spotify import SpotifyClient from resonance_engine.dsp.backends.spotify import SpotifyDSPBackend from resonance_engine.dsp.enums import DSPResource def _noisy() -> list[dict[str, Any]]: return [ { "id": "1", "name": "Item", "images": [{"url": "x"}], "external_urls": {"spotify": "u"}, "available_markets": ["US", "GB"], "album": {"id": "al", "name": "Album", "images": [{"url": "y"}]}, "tracks": [{"id": "t", "name": "T", "available_markets": ["US"]}], } ] def _clean() -> list[dict[str, Any]]: return [ { "id": "1", "name": "Item", "album": {"id": "al", "name": "Album"}, "tracks": [{"id": "t", "name": "T"}], } ] TOKEN = SecretStr("access-token") class TestSpotifyDSPBackend: @pytest.fixture def client_mock(self, mocker: MockerFixture) -> Any: return mocker.create_autospec(SpotifyClient, instance=True) @pytest.fixture def backend(self, client_mock: Any) -> SpotifyDSPBackend: return SpotifyDSPBackend(client_mock) def test_required_scope_maps_endpoints(self, backend: SpotifyDSPBackend) -> None: assert backend.required_scope(DSPResource.top_artists) == "user-top-read" assert ( backend.required_scope(DSPResource.recently_played) == "user-read-recently-played" ) assert backend.required_scope(DSPResource.profile) is None def test_has_required_scope(self, backend: SpotifyDSPBackend) -> None: granted = "user-top-read user-read-email" assert backend.has_required_scope(DSPResource.top_artists, granted) assert not backend.has_required_scope( DSPResource.top_artists, "user-read-email" ) # Unknown granted scopes or an unmapped resource → permissive (don't pre-skip). assert backend.has_required_scope(DSPResource.top_artists, None) assert backend.has_required_scope(DSPResource.profile, "") def test_refresh_token_returns_access_token( self, backend: SpotifyDSPBackend, client_mock: Any ) -> None: client_mock.refresh_token.return_value = {"access_token": SecretStr("at")} result = backend.refresh_token(SecretStr("refresh")) assert result == {"access_token": SecretStr("at")} client_mock.refresh_token.assert_called_once_with(SecretStr("refresh")) def test_refresh_token_includes_rotated_refresh_token( self, backend: SpotifyDSPBackend, client_mock: Any ) -> None: client_mock.refresh_token.return_value = { "access_token": SecretStr("at"), "refresh_token": SecretStr("rt2"), } result = backend.refresh_token(SecretStr("refresh")) assert result == { "access_token": SecretStr("at"), "refresh_token": SecretStr("rt2"), } def test_get_profile_passes_through( self, backend: SpotifyDSPBackend, client_mock: Any ) -> None: client_mock.get_current_user_profile.return_value = {"id": "u1"} assert backend.get_profile(TOKEN) == {"id": "u1"} client_mock.get_current_user_profile.assert_called_once_with(TOKEN) def test_close_closes_client( self, backend: SpotifyDSPBackend, client_mock: Any ) -> None: backend.close() client_mock.close.assert_called_once_with() @pytest.mark.parametrize( ("backend_method", "client_method", "max_items"), [ ("get_top_artists", "get_current_user_top_artists", 50), ("get_top_tracks", "get_current_user_top_tracks", 50), ("get_playlists", "get_current_user_playlists", 50), ("get_saved_albums", "get_current_user_saved_albums", 50), ("get_saved_tracks", "get_current_user_saved_tracks", 100), ("get_followed_artists", "get_current_user_followed_artists", 100), ], ) def test_strips_noise_keys_and_passes_max_items( self, backend: SpotifyDSPBackend, client_mock: Any, backend_method: str, client_method: str, max_items: int, ) -> None: getattr(client_mock, client_method).return_value = _noisy() result = getattr(backend, backend_method)(TOKEN) assert result == _clean() getattr(client_mock, client_method).assert_called_once_with( TOKEN, max_items=max_items ) def test_recently_played_passes_after_and_strips( self, backend: SpotifyDSPBackend, client_mock: Any ) -> None: client_mock.get_current_user_recently_played.return_value = _noisy() result = backend.get_recently_played(TOKEN, after=1700000000) assert result == _clean() client_mock.get_current_user_recently_played.assert_called_once_with( TOKEN, after=1700000000, max_items=200 )