from typing import Any import pytest from pydantic import SecretStr from pytest_mock import MockerFixture from resonance_engine.adapters.apple_music import AppleMusicClient from resonance_engine.dsp.backends.apple_music import AppleMusicDSPBackend from resonance_engine.dsp.enums import DSPId, DSPResource from resonance_engine.dsp.exceptions import DSPResourceUnsupportedError def _noisy() -> list[dict[str, Any]]: return [ { "id": "1", "type": "songs", "href": "/v1/catalog/us/songs/1", "attributes": { "name": "Item", "artwork": {"url": "x"}, "playParams": {"id": "1"}, "previews": [{"url": "x"}], "discNumber": 1, "trackNumber": 7, "hasLyrics": True, "isAppleDigitalMaster": True, }, } ] def _clean() -> list[dict[str, Any]]: return [ { "id": "1", "type": "songs", "attributes": {"name": "Item"}, } ] TOKEN = SecretStr("user-token") class TestAppleMusicDSPBackend: @pytest.fixture def client_mock(self, mocker: MockerFixture) -> Any: return mocker.create_autospec(AppleMusicClient, instance=True) @pytest.fixture def backend(self, client_mock: Any) -> AppleMusicDSPBackend: return AppleMusicDSPBackend(client_mock) def test_dsp_id(self, backend: AppleMusicDSPBackend) -> None: assert backend.dsp_id == DSPId.apple def test_required_scope_is_permissive(self, backend: AppleMusicDSPBackend) -> None: # MusicKit user tokens carry no OAuth scopes — nothing is pre-gated. assert backend.required_scope(DSPResource.playlists) is None assert backend.has_required_scope(DSPResource.saved_tracks, "anything") def test_close_closes_client( self, backend: AppleMusicDSPBackend, client_mock: Any ) -> None: backend.close() client_mock.close.assert_called_once_with() def test_refresh_token_echoes_user_token( self, backend: AppleMusicDSPBackend ) -> None: # Apple has no token exchange — the Music-User-Token is the access token. assert backend.refresh_token(SecretStr("user")) == { "access_token": SecretStr("user") } @pytest.mark.parametrize( ("backend_method", "client_method", "max_items"), [ ("get_recently_played", "get_recently_played_tracks", 30), ("get_playlists", "get_library_playlists", 100), ("get_saved_albums", "get_library_albums", 100), ("get_saved_tracks", "get_library_songs", 100), ("get_followed_artists", "get_library_artists", 100), ], ) def test_strips_noise_keys_and_passes_max_items( self, backend: AppleMusicDSPBackend, 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_ignores_after( self, backend: AppleMusicDSPBackend, client_mock: Any ) -> None: # Apple paginates by offset — the after cursor is accepted but unused. client_mock.get_recently_played_tracks.return_value = _noisy() result = backend.get_recently_played(TOKEN, after=1700000000) assert result == _clean() client_mock.get_recently_played_tracks.assert_called_once_with( TOKEN, max_items=30 ) @pytest.mark.parametrize( "backend_method", ["get_profile", "get_top_artists", "get_top_tracks"] ) def test_unsupported_resources_raise( self, backend: AppleMusicDSPBackend, backend_method: str ) -> None: with pytest.raises(DSPResourceUnsupportedError): getattr(backend, backend_method)(TOKEN)