import httpx import jwt import pytest import respx from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric import ec from fansifter_common.httpclient import HTTPClientError from pydantic import SecretStr from resonance_engine.adapters.apple_music import ( API_BASE, AppleMusicClient, AppleMusicHTTPError, ) def _es256_private_key() -> str: key = ec.generate_private_key(ec.SECP256R1()) return key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ).decode() class TestAppleMusicHTTPError: def test_detail_reads_error_detail(self) -> None: response = httpx.Response( 403, json={ "errors": [ {"status": "403", "title": "Forbidden", "detail": "No access"} ] }, ) exc = AppleMusicHTTPError("status 403", response=response) assert exc.detail == "No access" assert str(exc) == "status 403: No access" def test_detail_falls_back_to_title(self) -> None: response = httpx.Response( 403, json={"errors": [{"status": "403", "title": "Forbidden"}]} ) exc = AppleMusicHTTPError("status 403", response=response) assert exc.detail == "Forbidden" def test_detail_none_for_empty_errors(self) -> None: exc = AppleMusicHTTPError( "status 500", response=httpx.Response(500, json={"errors": []}) ) assert exc.detail is None assert str(exc) == "status 500" def test_detail_none_for_non_json_body(self) -> None: exc = AppleMusicHTTPError( "status 500", response=httpx.Response(500, content=b"nope") ) assert exc.detail is None def test_detail_none_without_response(self) -> None: exc = AppleMusicHTTPError("status 500") assert exc.detail is None assert str(exc) == "status 500" class TestAppleMusicClient: @pytest.fixture(scope="class") def client(self) -> AppleMusicClient: return AppleMusicClient( team_id="TEAM123456", key_id="KEY1234567", private_key=SecretStr(_es256_private_key()), ) def test_sends_developer_and_user_token_headers( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/recent/played/tracks").mock( return_value=httpx.Response(200, json={"data": [{"id": "t1"}]}) ) client.get_recently_played_tracks(SecretStr("user-tok")) headers = respx_mock.calls[0].request.headers assert headers["music-user-token"] == "user-tok" bearer = headers["authorization"].removeprefix("Bearer ") claims = jwt.decode(bearer, options={"verify_signature": False}) assert claims["iss"] == "TEAM123456" header = jwt.get_unverified_header(bearer) assert header["kid"] == "KEY1234567" assert header["alg"] == "ES256" def test_developer_token_is_cached_across_requests( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/recent/played/tracks").mock( return_value=httpx.Response(200, json={"data": [{"id": "t1"}]}) ) client.get_recently_played_tracks(SecretStr("user-tok")) client.get_recently_played_tracks(SecretStr("user-tok")) first = respx_mock.calls[0].request.headers["authorization"] second = respx_mock.calls[1].request.headers["authorization"] assert first == second def test_get_recently_played_returns_items_on_single_page( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: track = {"id": "t1", "type": "songs"} respx_mock.get(f"{API_BASE}/me/recent/played/tracks").mock( return_value=httpx.Response(200, json={"data": [track]}) ) result = client.get_recently_played_tracks(SecretStr("user-tok")) assert result == [track] def test_get_recently_played_skips_null_items( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: track = {"id": "t1", "type": "songs"} respx_mock.get(f"{API_BASE}/me/recent/played/tracks").mock( return_value=httpx.Response(200, json={"data": [None, track]}) ) result = client.get_recently_played_tracks(SecretStr("user-tok")) assert result == [track] def test_get_library_songs_follows_offset_across_pages( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: song1 = {"id": "s1", "type": "library-songs"} song2 = {"id": "s2", "type": "library-songs"} def side_effect(request: httpx.Request) -> httpx.Response: offset = int(request.url.params.get("offset", 0)) if offset == 0: return httpx.Response( 200, json={"data": [song1], "next": "/v1/me/library/songs?offset=1"} ) return httpx.Response(200, json={"data": [song2]}) respx_mock.get(f"{API_BASE}/me/library/songs").mock(side_effect=side_effect) result = client.get_library_songs(SecretStr("user-tok")) assert result == [song1, song2] assert respx_mock.calls[1].request.url.params["offset"] == "1" def test_get_library_songs_stops_when_no_next( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: song = {"id": "s1", "type": "library-songs"} respx_mock.get(f"{API_BASE}/me/library/songs").mock( return_value=httpx.Response(200, json={"data": [song]}) ) result = client.get_library_songs(SecretStr("user-tok")) assert result == [song] assert len(respx_mock.calls) == 1 def test_get_library_albums_truncates_to_max_items( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: albums = [{"id": f"a{i}", "type": "library-albums"} for i in range(3)] respx_mock.get(f"{API_BASE}/me/library/albums").mock( return_value=httpx.Response( 200, json={"data": albums, "next": "/v1/me/library/albums?offset=3"} ) ) result = client.get_library_albums(SecretStr("user-tok"), max_items=2) assert result == albums[:2] assert respx_mock.calls[0].request.url.params["limit"] == "2" def test_get_library_playlists_returns_items( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: playlist = {"id": "p1", "type": "library-playlists"} respx_mock.get(f"{API_BASE}/me/library/playlists").mock( return_value=httpx.Response(200, json={"data": [playlist]}) ) result = client.get_library_playlists(SecretStr("user-tok")) assert result == [playlist] def test_get_library_artists_returns_items( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: artist = {"id": "ar1", "type": "library-artists"} respx_mock.get(f"{API_BASE}/me/library/artists").mock( return_value=httpx.Response(200, json={"data": [artist]}) ) result = client.get_library_artists(SecretStr("user-tok")) assert result == [artist] def test_raises_api_error_on_401( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/recent/played/tracks").mock( return_value=httpx.Response(401) ) with pytest.raises(HTTPClientError) as exc: client.get_recently_played_tracks(SecretStr("user-tok")) assert exc.value.status_code == 401 def test_get_library_songs_raises_api_error_on_network_error( self, client: AppleMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/library/songs").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(HTTPClientError): client.get_library_songs(SecretStr("user-tok"))