from typing import Any import pytest from pydantic import SecretStr from pytest_mock import MockerFixture from resonance_engine.adapters.amazon_music import AmazonMusicClient from resonance_engine.dsp.backends.amazon_music import AmazonMusicDSPBackend from resonance_engine.dsp.enums import DSPId, DSPResource from resonance_engine.dsp.exceptions import DSPResourceUnsupportedError def _noisy() -> list[dict[str, Any]]: return [ { "id": "1", "title": "Item", "isrc": "US123", "url": "https://music.amazon.com/tracks/1", "visibility": "PUBLIC", "playbackVisibility": "PRIVATE", "playParams": {"id": "mrn:..."}, "playbackAvailability": {"isAllowed": True}, "eligibility": {"isPrimeEligible": False}, "images": [{"url": "x"}], "art": {"url": "y"}, "album": {"id": "al", "title": "Album", "artwork": {"url": "z"}}, } ] def _clean() -> list[dict[str, Any]]: return [ { "id": "1", "title": "Item", "isrc": "US123", "album": {"id": "al", "title": "Album"}, } ] TOKEN = SecretStr("access-token") class TestAmazonMusicDSPBackend: @pytest.fixture def client_mock(self, mocker: MockerFixture) -> Any: return mocker.create_autospec(AmazonMusicClient, instance=True) @pytest.fixture def backend(self, client_mock: Any) -> AmazonMusicDSPBackend: return AmazonMusicDSPBackend(client_mock) def test_dsp_id(self, backend: AmazonMusicDSPBackend) -> None: assert backend.dsp_id == DSPId.amazon def test_required_scope_is_permissive(self, backend: AmazonMusicDSPBackend) -> None: # Beta API scopes are undocumented — no resource is pre-gated on a scope. assert backend.required_scope(DSPResource.top_artists) is None assert backend.has_required_scope(DSPResource.playlists, "anything") def test_close_closes_client( self, backend: AmazonMusicDSPBackend, client_mock: Any ) -> None: backend.close() client_mock.close.assert_called_once_with() def test_refresh_token_passes_secret_and_returns_access_token( self, backend: AmazonMusicDSPBackend, 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: AmazonMusicDSPBackend, 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_refresh_token_omits_null_refresh_token( self, backend: AmazonMusicDSPBackend, client_mock: Any ) -> None: client_mock.refresh_token.return_value = { "access_token": SecretStr("at"), "refresh_token": None, } result = backend.refresh_token(SecretStr("refresh")) assert result == {"access_token": SecretStr("at")} def test_get_profile_strips_noise( self, backend: AmazonMusicDSPBackend, client_mock: Any ) -> None: client_mock.get_current_user_profile.return_value = _noisy()[0] assert backend.get_profile(TOKEN) == _clean()[0] client_mock.get_current_user_profile.assert_called_once_with(TOKEN) @pytest.mark.parametrize( ("backend_method", "client_method", "max_items"), [ ("get_recently_played", "get_current_user_recently_played", 60), ("get_playlists", "get_current_user_playlists", 100), ("get_followed_artists", "get_current_user_followed_artists", 100), ], ) def test_strips_noise_keys_and_passes_max_items( self, backend: AmazonMusicDSPBackend, 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 ) @pytest.mark.parametrize( "backend_method", ["get_top_artists", "get_top_tracks", "get_saved_albums", "get_saved_tracks"], ) def test_unsupported_resources_raise( self, backend: AmazonMusicDSPBackend, backend_method: str, ) -> None: with pytest.raises(DSPResourceUnsupportedError): getattr(backend, backend_method)(TOKEN)