import base64 import httpx import pytest import respx from fansifter_common.httpclient import HTTPClientError from pydantic import SecretStr from resonance_engine.adapters.amazon_music import ( API_BASE, TOKEN_URL, AmazonMusicClient, AmazonMusicHTTPError, AmazonMusicTokenError, AmazonMusicTokenRevokedError, ) class TestAmazonMusicHTTPError: def test_detail_appends_api_error_object_message(self) -> None: response = httpx.Response(403, json={"error": {"message": "Forbidden"}}) exc = AmazonMusicHTTPError("status 403", response=response) assert exc.detail == "Forbidden" assert str(exc) == "status 403: Forbidden" def test_detail_reads_string_error(self) -> None: response = httpx.Response(400, json={"error": "invalid_request"}) exc = AmazonMusicHTTPError("status 400", response=response) assert exc.detail == "invalid_request" assert str(exc) == "status 400: invalid_request" def test_detail_falls_back_to_top_level_message(self) -> None: response = httpx.Response(404, json={"message": "Not found"}) exc = AmazonMusicHTTPError("status 404", response=response) assert exc.detail == "Not found" def test_detail_none_for_non_json_body(self) -> None: exc = AmazonMusicHTTPError( "status 500", response=httpx.Response(500, content=b"nope") ) assert exc.detail is None assert str(exc) == "status 500" def test_detail_none_without_response(self) -> None: exc = AmazonMusicHTTPError("status 500") assert exc.detail is None assert str(exc) == "status 500" class TestAmazonMusicClientRefreshToken: @pytest.fixture(scope="class") def client(self) -> AmazonMusicClient: return AmazonMusicClient("id", SecretStr("secret"), "amzn1.application.key") def test_refresh_token_returns_token( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response( 200, json={ "access_token": "acc", "refresh_token": "newref", "token_type": "bearer", "expires_in": 3600, }, ) ) result = client.refresh_token(SecretStr("ref")) assert result == { "access_token": SecretStr("acc"), "refresh_token": SecretStr("newref"), "token_type": "bearer", "expires_in": 3600, } def test_refresh_token_sends_grant_in_body_and_creds_via_basic_auth( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response(200, json={"access_token": "acc"}) ) client.refresh_token(SecretStr("ref")) request = respx_mock.calls[0].request auth = request.headers["authorization"].removeprefix("Basic ") assert base64.b64decode(auth).decode() == "id:secret" content = request.content.decode() assert "grant_type=refresh_token" in content assert "refresh_token=ref" in content assert "client_secret" not in content def test_refresh_token_raises_revoked_on_invalid_grant( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response( 400, json={ "error": "invalid_grant", "error_description": "Refresh token revoked", }, ) ) with pytest.raises(AmazonMusicTokenRevokedError, match="Refresh token revoked"): client.refresh_token(SecretStr("ref")) def test_refresh_token_raises_revoked_on_unauthorized_client( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response( 400, json={"error": "unauthorized_client"}, ) ) with pytest.raises(AmazonMusicTokenRevokedError, match="unauthorized_client"): client.refresh_token(SecretStr("ref")) def test_refresh_token_raises_transient_on_non_revoked_oauth_error( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response( 400, json={ "error": "invalid_client", "error_description": "Invalid credentials", }, ) ) with pytest.raises(AmazonMusicTokenError, match="Invalid credentials") as exc: client.refresh_token(SecretStr("ref")) assert not isinstance(exc.value, AmazonMusicTokenRevokedError) def test_refresh_token_raises_transient_on_non_json_error( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock(return_value=httpx.Response(500)) with pytest.raises(AmazonMusicTokenError, match="HTTP 500") as exc: client.refresh_token(SecretStr("ref")) assert not isinstance(exc.value, AmazonMusicTokenRevokedError) def test_refresh_token_raises_transient_on_network_error( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock(side_effect=httpx.ConnectError("unreachable")) with pytest.raises(AmazonMusicTokenError): client.refresh_token(SecretStr("ref")) class TestAmazonMusicClientProfile: @pytest.fixture(scope="class") def client(self) -> AmazonMusicClient: return AmazonMusicClient("id", SecretStr("secret"), "amzn1.application.key") def test_get_profile_unwraps_data_user( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: user = {"id": "u1", "displayName": "Fan"} respx_mock.get(f"{API_BASE}/me").mock( return_value=httpx.Response(200, json={"data": {"user": user}}) ) result = client.get_current_user_profile(SecretStr("acc")) assert result == user def test_get_profile_falls_back_to_raw_when_unwrapped( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: profile = {"id": "u1", "displayName": "Fan"} respx_mock.get(f"{API_BASE}/me").mock( return_value=httpx.Response(200, json=profile) ) result = client.get_current_user_profile(SecretStr("acc")) assert result == profile def test_get_profile_sends_bearer_and_api_key_headers( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me").mock( return_value=httpx.Response(200, json={"data": {"user": {}}}) ) client.get_current_user_profile(SecretStr("acc")) headers = respx_mock.calls[0].request.headers assert headers["authorization"] == "Bearer acc" assert headers["x-api-key"] == "amzn1.application.key" def test_get_profile_raises_api_error_on_401( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me").mock(return_value=httpx.Response(401)) with pytest.raises(HTTPClientError) as exc: client.get_current_user_profile(SecretStr("acc")) assert exc.value.status_code == 401 def test_get_profile_raises_api_error_on_network_error( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me").mock(side_effect=httpx.ConnectError("timeout")) with pytest.raises(HTTPClientError): client.get_current_user_profile(SecretStr("acc")) def _connection( edges: list, *, has_next: bool = False, token: str | None = None ) -> dict: return { "edges": edges, "pageInfo": {"hasNextPage": has_next, "token": token}, } class TestAmazonMusicClientRecentlyPlayed: @pytest.fixture(scope="class") def client(self) -> AmazonMusicClient: return AmazonMusicClient("id", SecretStr("secret"), "amzn1.application.key") def _respond(self, connection: dict) -> httpx.Response: return httpx.Response( 200, json={"data": {"user": {"recentTrackPlayback": connection}}}, ) def test_merges_node_with_played_at_on_single_page( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: edge = { "node": {"id": "t1", "title": "Track One"}, "playedAt": "2024-01-01T12:00:00Z", } respx_mock.get(f"{API_BASE}/player/recentlyPlayed").mock( return_value=self._respond(_connection([edge])) ) result = client.get_current_user_recently_played(SecretStr("acc")) assert result == [ {"id": "t1", "title": "Track One", "playedAt": "2024-01-01T12:00:00Z"} ] def test_skips_edges_without_node( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: edges = [ {"node": None, "playedAt": "2024-01-01T12:00:00Z"}, {"node": {"id": "t1"}, "playedAt": "2024-01-02T12:00:00Z"}, ] respx_mock.get(f"{API_BASE}/player/recentlyPlayed").mock( return_value=self._respond(_connection(edges)) ) result = client.get_current_user_recently_played(SecretStr("acc")) assert result == [{"id": "t1", "playedAt": "2024-01-02T12:00:00Z"}] def test_follows_cursor_across_pages( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: edge1 = {"node": {"id": "t1"}, "playedAt": "2024-01-01T00:00:00Z"} edge2 = {"node": {"id": "t2"}, "playedAt": "2024-01-02T00:00:00Z"} def side_effect(request: httpx.Request) -> httpx.Response: if request.url.params.get("cursor") is None: return self._respond(_connection([edge1], has_next=True, token="c2")) return self._respond(_connection([edge2])) respx_mock.get(f"{API_BASE}/player/recentlyPlayed").mock( side_effect=side_effect ) result = client.get_current_user_recently_played(SecretStr("acc")) assert result == [ {"id": "t1", "playedAt": "2024-01-01T00:00:00Z"}, {"id": "t2", "playedAt": "2024-01-02T00:00:00Z"}, ] assert respx_mock.calls[1].request.url.params["cursor"] == "c2" def test_stops_when_next_page_flagged_but_token_missing( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: edge = {"node": {"id": "t1"}, "playedAt": "2024-01-01T00:00:00Z"} respx_mock.get(f"{API_BASE}/player/recentlyPlayed").mock( return_value=self._respond(_connection([edge], has_next=True, token=None)) ) result = client.get_current_user_recently_played(SecretStr("acc")) assert result == [{"id": "t1", "playedAt": "2024-01-01T00:00:00Z"}] assert len(respx_mock.calls) == 1 def test_truncates_to_max_items( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: edges = [ {"node": {"id": f"t{i}"}, "playedAt": f"2024-01-0{i}T00:00:00Z"} for i in range(1, 4) ] respx_mock.get(f"{API_BASE}/player/recentlyPlayed").mock( return_value=self._respond(_connection(edges, has_next=True, token="c2")) ) result = client.get_current_user_recently_played(SecretStr("acc"), max_items=2) assert result == [ {"id": "t1", "playedAt": "2024-01-01T00:00:00Z"}, {"id": "t2", "playedAt": "2024-01-02T00:00:00Z"}, ] assert respx_mock.calls[0].request.url.params["limit"] == "2" def test_raises_api_error_on_network_error( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/player/recentlyPlayed").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(HTTPClientError): client.get_current_user_recently_played(SecretStr("acc")) class TestAmazonMusicClientCollections: @pytest.fixture(scope="class") def client(self) -> AmazonMusicClient: return AmazonMusicClient("id", SecretStr("secret"), "amzn1.application.key") def test_get_playlists_returns_nodes_on_single_page( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: playlist = {"id": "p1", "name": "Playlist One"} respx_mock.get(f"{API_BASE}/me/playlists").mock( return_value=httpx.Response( 200, json={ "data": {"user": {"playlists": _connection([{"node": playlist}])}} }, ) ) result = client.get_current_user_playlists(SecretStr("acc")) assert result == [playlist] def test_get_playlists_follows_cursor_across_pages( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: playlist1 = {"id": "p1", "name": "Playlist One"} playlist2 = {"id": "p2", "name": "Playlist Two"} def side_effect(request: httpx.Request) -> httpx.Response: if request.url.params.get("cursor") is None: conn = _connection([{"node": playlist1}], has_next=True, token="c2") else: conn = _connection([{"node": playlist2}]) return httpx.Response(200, json={"data": {"user": {"playlists": conn}}}) respx_mock.get(f"{API_BASE}/me/playlists").mock(side_effect=side_effect) result = client.get_current_user_playlists(SecretStr("acc")) assert result == [playlist1, playlist2] def test_get_followed_artists_returns_nodes( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: artist = {"id": "a1", "name": "Artist One"} respx_mock.get(f"{API_BASE}/me/followed/artists").mock( return_value=httpx.Response( 200, json={ "data": { "user": {"followedArtists": _connection([{"node": artist}])} } }, ) ) result = client.get_current_user_followed_artists(SecretStr("acc")) assert result == [artist] def test_get_followed_artists_raises_api_error_on_401( self, client: AmazonMusicClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/followed/artists").mock( return_value=httpx.Response(401) ) with pytest.raises(HTTPClientError) as exc: client.get_current_user_followed_artists(SecretStr("acc")) assert exc.value.status_code == 401