import httpx import pytest import respx from app.adapters.spotify import ( API_BASE, TOKEN_URL, SpotifyAPIError, SpotifyClient, SpotifyTokenError, SpotifyTokenRevokedError, ) def _playlist(pid: str) -> dict: return { "id": pid, "name": f"Playlist {pid}", "description": None, "public": True, "collaborative": False, "snapshot_id": f"snap_{pid}", "type": "playlist", "uri": f"spotify:playlist:{pid}", "tracks": {"href": f"{API_BASE}/playlists/{pid}/tracks", "total": 5}, } def _playlists_page( items: list[dict], *, offset: int = 0, total: int | None = None, next_url: str | None = None, ) -> dict: return { "items": items, "total": total if total is not None else len(items), "limit": 50, "offset": offset, "href": f"{API_BASE}/me/playlists", "next": next_url, } def _saved_album(aid: str) -> dict: return { "added_at": "2024-01-01T00:00:00Z", "album": { "id": aid, "name": f"Album {aid}", "album_type": "album", "total_tracks": 10, "release_date": "2024-01-01", "release_date_precision": "day", "type": "album", "uri": f"spotify:album:{aid}", "artists": [ { "id": "ar1", "name": "Artist One", "type": "artist", "uri": "spotify:artist:ar1", } ], "label": "Some Label", "popularity": 75, }, } def _saved_albums_page( items: list[dict], *, offset: int = 0, total: int | None = None, next_url: str | None = None, ) -> dict: return { "items": items, "total": total if total is not None else len(items), "limit": 50, "offset": offset, "href": f"{API_BASE}/me/albums", "next": next_url, } class TestSpotifyClient: @pytest.fixture(scope="class") def client(self) -> SpotifyClient: return SpotifyClient("id", "secret") def test_refresh_token_returns_token( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response( 200, json={ "access_token": "acc", "token_type": "Bearer", "scope": "user-read-private", "expires_in": 3600, }, ) ) result = client.refresh_token("ref") assert result.access_token == "acc" assert result.expires_in == 3600 def test_refresh_token_raises_revoked_on_invalid_grant( self, client: SpotifyClient, 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(SpotifyTokenRevokedError, match="Refresh token revoked"): client.refresh_token("ref") def test_refresh_token_raises_revoked_on_unauthorized_client( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock( return_value=httpx.Response( 400, json={ "error": "unauthorized_client", "error_description": "Client not authorized", }, ) ) with pytest.raises(SpotifyTokenRevokedError, match="Client not authorized"): client.refresh_token("ref") def test_refresh_token_raises_token_error_on_invalid_client( self, client: SpotifyClient, 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(SpotifyTokenError, match="Invalid credentials"): client.refresh_token("ref") def test_refresh_token_raises_token_error_on_non_json_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock(return_value=httpx.Response(500)) with pytest.raises(SpotifyTokenError, match="HTTP 500"): client.refresh_token("ref") def test_refresh_token_raises_token_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.post(TOKEN_URL).mock(side_effect=httpx.ConnectError("unreachable")) with pytest.raises(SpotifyTokenError): client.refresh_token("ref") def test_get_profile_returns_profile( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me").mock( return_value=httpx.Response( 200, json={ "id": "user123", "email": "user@example.com", "display_name": "Test User", }, ) ) result = client.get_current_user_profile("acc") assert result.id == "user123" assert result.email == "user@example.com" def test_get_profile_raises_api_error_on_401( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me").mock(return_value=httpx.Response(401)) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_profile("acc") assert exc_info.value.status_code == 401 def test_get_profile_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me").mock(side_effect=httpx.ConnectError("timeout")) with pytest.raises(SpotifyAPIError): client.get_current_user_profile("acc") def test_get_top_artists_returns_all_items_on_single_page( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/artists").mock( return_value=httpx.Response( 200, json={ "items": [ { "id": "a1", "name": "Artist One", "popularity": 80, "genres": ["pop"], "type": "artist", "uri": "spotify:artist:a1", } ], "total": 1, "limit": 50, "offset": 0, "href": f"{API_BASE}/me/top/artists", }, ) ) result = client.get_current_user_top_artists("acc") assert len(result) == 1 assert result[0].id == "a1" assert result[0].name == "Artist One" def test_get_top_artists_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: def side_effect(request: httpx.Request) -> httpx.Response: offset = int(request.url.params.get("offset", 0)) if offset == 0: return httpx.Response( 200, json={ "items": [ { "id": "a1", "name": "Artist One", "popularity": 80, "genres": ["pop"], "type": "artist", "uri": "spotify:artist:a1", } ], "total": 2, "limit": 50, "offset": 0, "href": f"{API_BASE}/me/top/artists", "next": f"{API_BASE}/me/top/artists?offset=50", }, ) return httpx.Response( 200, json={ "items": [ { "id": "a2", "name": "Artist Two", "popularity": 60, "genres": ["rock"], "type": "artist", "uri": "spotify:artist:a2", } ], "total": 2, "limit": 50, "offset": 50, "href": f"{API_BASE}/me/top/artists", }, ) respx_mock.get(f"{API_BASE}/me/top/artists").mock(side_effect=side_effect) result = client.get_current_user_top_artists("acc") assert len(result) == 2 assert result[0].id == "a1" assert result[1].id == "a2" def test_get_top_artists_passes_time_range( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/artists").mock( return_value=httpx.Response( 200, json={ "items": [], "total": 0, "limit": 50, "offset": 0, "href": f"{API_BASE}/me/top/artists", }, ) ) client.get_current_user_top_artists("acc", time_range="long_term") assert respx_mock.calls[0].request.url.params["time_range"] == "long_term" def test_get_top_artists_raises_api_error_with_retry_after_on_429( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/artists").mock( return_value=httpx.Response(429, headers={"Retry-After": "30"}) ) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_top_artists("acc") assert exc_info.value.status_code == 429 assert exc_info.value.retry_after == 30 def test_get_top_artists_raises_api_error_without_retry_after_on_429( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/artists").mock( return_value=httpx.Response(429) ) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_top_artists("acc") assert exc_info.value.status_code == 429 assert exc_info.value.retry_after is None def test_get_top_tracks_returns_all_items_on_single_page( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: artist = { "id": "ar1", "name": "Artist One", "type": "artist", "uri": "spotify:artist:ar1", } album = { "id": "al1", "name": "Some Album", "album_type": "album", "release_date": "2024-01-01", "release_date_precision": "day", "type": "album", "uri": "spotify:album:al1", "artists": [artist], } respx_mock.get(f"{API_BASE}/me/top/tracks").mock( return_value=httpx.Response( 200, json={ "items": [ { "id": "t1", "name": "Track One", "popularity": 75, "type": "track", "uri": "spotify:track:t1", "duration_ms": 200000, "explicit": False, "track_number": 1, "disc_number": 1, "album": album, "artists": [artist], } ], "total": 1, "limit": 50, "offset": 0, "href": f"{API_BASE}/me/top/tracks", }, ) ) result = client.get_current_user_top_tracks("acc") assert len(result) == 1 assert result[0].id == "t1" assert result[0].name == "Track One" def test_get_top_tracks_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: artist = { "id": "ar1", "name": "Artist One", "type": "artist", "uri": "spotify:artist:ar1", } album = { "id": "al1", "name": "Some Album", "album_type": "album", "release_date": "2024-01-01", "release_date_precision": "day", "type": "album", "uri": "spotify:album:al1", "artists": [artist], } def side_effect(request: httpx.Request) -> httpx.Response: offset = int(request.url.params.get("offset", 0)) track_id = "t1" if offset == 0 else "t2" return httpx.Response( 200, json={ "items": [ { "id": track_id, "name": f"Track {track_id}", "popularity": 75, "type": "track", "uri": f"spotify:track:{track_id}", "duration_ms": 200000, "explicit": False, "track_number": 1, "disc_number": 1, "album": album, "artists": [artist], } ], "total": 2, "limit": 50, "offset": offset, "href": f"{API_BASE}/me/top/tracks", "next": f"{API_BASE}/me/top/tracks?offset=50" if offset == 0 else None, }, ) respx_mock.get(f"{API_BASE}/me/top/tracks").mock(side_effect=side_effect) result = client.get_current_user_top_tracks("acc") assert len(result) == 2 assert result[0].id == "t1" assert result[1].id == "t2" def test_get_top_tracks_passes_time_range( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/tracks").mock( return_value=httpx.Response( 200, json={ "items": [], "total": 0, "limit": 50, "offset": 0, "href": f"{API_BASE}/me/top/tracks", }, ) ) client.get_current_user_top_tracks("acc", time_range="short_term") assert respx_mock.calls[0].request.url.params["time_range"] == "short_term" def test_get_top_tracks_raises_api_error_on_429( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/tracks").mock( return_value=httpx.Response(429, headers={"Retry-After": "10"}) ) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_top_tracks("acc") assert exc_info.value.status_code == 429 assert exc_info.value.retry_after == 10 def test_get_top_tracks_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/top/tracks").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(SpotifyAPIError): client.get_current_user_top_tracks("acc") def test_get_recently_played_returns_recently_played( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/player/recently-played").mock( return_value=httpx.Response( 200, json={ "items": [ { "track": { "id": "t1", "name": "Track One", "popularity": 70, "type": "track", "uri": "spotify:track:t1", }, "played_at": "2024-01-01T12:00:00Z", } ], "limit": 50, "href": f"{API_BASE}/me/player/recently-played", }, ) ) result = client.get_current_user_recently_played("acc") assert len(result) == 1 assert result[0].track.id == "t1" assert result[0].played_at == "2024-01-01T12:00:00Z" def test_get_recently_played_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: def side_effect(request: httpx.Request) -> httpx.Response: after = request.url.params.get("after") if after is None: return httpx.Response( 200, json={ "items": [ { "track": { "id": "t1", "name": "Track One", "popularity": 70, "type": "track", "uri": "spotify:track:t1", }, "played_at": "2024-01-01T12:00:00Z", } ], "limit": 50, "href": f"{API_BASE}/me/player/recently-played", "next": f"{API_BASE}/me/player/recently-played?after=1704067200000", "cursors": { "after": "1704067200000", "before": "1704067200001", }, }, ) return httpx.Response( 200, json={ "items": [ { "track": { "id": "t2", "name": "Track Two", "popularity": 60, "type": "track", "uri": "spotify:track:t2", }, "played_at": "2024-01-02T12:00:00Z", } ], "limit": 50, "href": f"{API_BASE}/me/player/recently-played", }, ) respx_mock.get(f"{API_BASE}/me/player/recently-played").mock( side_effect=side_effect ) result = client.get_current_user_recently_played("acc") assert len(result) == 2 assert result[0].track.id == "t1" assert result[1].track.id == "t2" def test_get_recently_played_passes_after_parameter( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/player/recently-played").mock( return_value=httpx.Response( 200, json={ "items": [], "limit": 50, "href": f"{API_BASE}/me/player/recently-played", }, ) ) client.get_current_user_recently_played("acc", after=1704067200000) assert respx_mock.calls[0].request.url.params["after"] == "1704067200000" def test_get_recently_played_raises_api_error_on_403( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/player/recently-played").mock( return_value=httpx.Response(403) ) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_recently_played("acc") assert exc_info.value.status_code == 403 def test_get_recently_played_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/player/recently-played").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(SpotifyAPIError): client.get_current_user_recently_played("acc") def test_get_current_user_playlists_returns_all_items_on_single_page( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/playlists").mock( return_value=httpx.Response( 200, json=_playlists_page([_playlist("p1"), _playlist("p2")]), ) ) result = client.get_current_user_playlists("acc") assert len(result) == 2 assert result[0].id == "p1" assert result[1].id == "p2" def test_get_current_user_playlists_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: next_url = f"{API_BASE}/me/playlists?offset=50&limit=50" def side_effect(request: httpx.Request) -> httpx.Response: offset = int(request.url.params.get("offset", 0)) if offset == 0: return httpx.Response( 200, json=_playlists_page([_playlist("p1")], total=2, next_url=next_url), ) return httpx.Response( 200, json=_playlists_page([_playlist("p2")], offset=50, total=2), ) respx_mock.get(f"{API_BASE}/me/playlists").mock(side_effect=side_effect) result = client.get_current_user_playlists("acc") assert len(result) == 2 assert result[0].id == "p1" assert result[1].id == "p2" def test_get_current_user_playlists_raises_api_error_on_401( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/playlists").mock( return_value=httpx.Response(401) ) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_playlists("acc") assert exc_info.value.status_code == 401 def test_get_current_user_playlists_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/playlists").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(SpotifyAPIError): client.get_current_user_playlists("acc") def test_get_saved_albums_returns_all_items_on_single_page( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/albums").mock( return_value=httpx.Response( 200, json=_saved_albums_page([_saved_album("a1"), _saved_album("a2")]), ) ) result = client.get_current_user_saved_albums("acc") assert len(result) == 2 assert result[0].album.id == "a1" assert result[1].album.id == "a2" def test_get_saved_albums_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: next_url = f"{API_BASE}/me/albums?offset=50&limit=50" def side_effect(request: httpx.Request) -> httpx.Response: offset = int(request.url.params.get("offset", 0)) if offset == 0: return httpx.Response( 200, json=_saved_albums_page( [_saved_album("a1")], total=2, next_url=next_url ), ) return httpx.Response( 200, json=_saved_albums_page([_saved_album("a2")], offset=50, total=2), ) respx_mock.get(f"{API_BASE}/me/albums").mock(side_effect=side_effect) result = client.get_current_user_saved_albums("acc") assert len(result) == 2 assert result[0].album.id == "a1" assert result[1].album.id == "a2" def test_get_saved_albums_raises_api_error_on_401( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/albums").mock(return_value=httpx.Response(401)) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_saved_albums("acc") assert exc_info.value.status_code == 401 def test_get_saved_albums_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/albums").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(SpotifyAPIError): client.get_current_user_saved_albums("acc") def test_get_saved_tracks_returns_all_items_on_single_page( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: artist = { "id": "ar1", "name": "Artist One", "type": "artist", "uri": "spotify:artist:ar1", } album = { "id": "al1", "name": "Some Album", "album_type": "album", "release_date": "2024-01-01", "release_date_precision": "day", "type": "album", "uri": "spotify:album:al1", "artists": [artist], } respx_mock.get(f"{API_BASE}/me/tracks").mock( return_value=httpx.Response( 200, json={ "items": [ { "added_at": "2024-01-01T00:00:00Z", "track": { "id": "t1", "name": "Track t1", "popularity": 70, "type": "track", "uri": "spotify:track:t1", "duration_ms": 210000, "explicit": False, "track_number": 1, "disc_number": 1, "album": album, "artists": [artist], }, }, { "added_at": "2024-01-01T00:00:00Z", "track": { "id": "t2", "name": "Track t2", "popularity": 70, "type": "track", "uri": "spotify:track:t2", "duration_ms": 210000, "explicit": False, "track_number": 2, "disc_number": 1, "album": album, "artists": [artist], }, }, ], "total": 2, "limit": 50, "offset": 0, "href": f"{API_BASE}/me/tracks", }, ) ) result = client.get_current_user_saved_tracks("acc") assert len(result) == 2 assert result[0].track.id == "t1" assert result[1].track.id == "t2" def test_get_saved_tracks_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: artist = { "id": "ar1", "name": "Artist One", "type": "artist", "uri": "spotify:artist:ar1", } album = { "id": "al1", "name": "Some Album", "album_type": "album", "release_date": "2024-01-01", "release_date_precision": "day", "type": "album", "uri": "spotify:album:al1", "artists": [artist], } next_url = f"{API_BASE}/me/tracks?offset=50&limit=50" def side_effect(request: httpx.Request) -> httpx.Response: offset = int(request.url.params.get("offset", 0)) track_id = "t1" if offset == 0 else "t2" return httpx.Response( 200, json={ "items": [ { "added_at": "2024-01-01T00:00:00Z", "track": { "id": track_id, "name": f"Track {track_id}", "popularity": 70, "type": "track", "uri": f"spotify:track:{track_id}", "duration_ms": 210000, "explicit": False, "track_number": 1, "disc_number": 1, "album": album, "artists": [artist], }, } ], "total": 2, "limit": 50, "offset": offset, "href": f"{API_BASE}/me/tracks", "next": next_url if offset == 0 else None, }, ) respx_mock.get(f"{API_BASE}/me/tracks").mock(side_effect=side_effect) result = client.get_current_user_saved_tracks("acc") assert len(result) == 2 assert result[0].track.id == "t1" assert result[1].track.id == "t2" def test_get_saved_tracks_raises_api_error_on_401( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/tracks").mock(return_value=httpx.Response(401)) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_saved_tracks("acc") assert exc_info.value.status_code == 401 def test_get_saved_tracks_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/tracks").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(SpotifyAPIError): client.get_current_user_saved_tracks("acc") def test_get_followed_artists_returns_all_items_on_single_page( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/following").mock( return_value=httpx.Response( 200, json={ "artists": { "items": [ { "id": "a1", "name": "Artist One", "popularity": 80, "genres": ["pop"], "type": "artist", "uri": "spotify:artist:a1", }, { "id": "a2", "name": "Artist Two", "popularity": 60, "genres": ["rock"], "type": "artist", "uri": "spotify:artist:a2", }, ], "total": 2, "limit": 50, "href": f"{API_BASE}/me/following?type=artist", "next": None, } }, ) ) result = client.get_current_user_followed_artists("acc") assert len(result) == 2 assert result[0].id == "a1" assert result[1].id == "a2" def test_get_followed_artists_fetches_all_pages( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: def side_effect(request: httpx.Request) -> httpx.Response: after = request.url.params.get("after") if after is None: return httpx.Response( 200, json={ "artists": { "items": [ { "id": "a1", "name": "Artist One", "popularity": 80, "genres": ["pop"], "type": "artist", "uri": "spotify:artist:a1", } ], "total": 2, "limit": 50, "href": f"{API_BASE}/me/following?type=artist", "next": f"{API_BASE}/me/following?type=artist&after=a1", "cursors": {"after": "a1", "before": ""}, } }, ) return httpx.Response( 200, json={ "artists": { "items": [ { "id": "a2", "name": "Artist Two", "popularity": 60, "genres": ["rock"], "type": "artist", "uri": "spotify:artist:a2", } ], "total": 2, "limit": 50, "href": f"{API_BASE}/me/following?type=artist", "next": None, } }, ) respx_mock.get(f"{API_BASE}/me/following").mock(side_effect=side_effect) result = client.get_current_user_followed_artists("acc") assert len(result) == 2 assert result[0].id == "a1" assert result[1].id == "a2" def test_get_followed_artists_raises_api_error_on_401( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/following").mock( return_value=httpx.Response(401) ) with pytest.raises(SpotifyAPIError) as exc_info: client.get_current_user_followed_artists("acc") assert exc_info.value.status_code == 401 def test_get_followed_artists_raises_api_error_on_network_error( self, client: SpotifyClient, respx_mock: respx.MockRouter ) -> None: respx_mock.get(f"{API_BASE}/me/following").mock( side_effect=httpx.ConnectError("timeout") ) with pytest.raises(SpotifyAPIError): client.get_current_user_followed_artists("acc")