from __future__ import annotations import httpx2 import pytest import spotify_mcp_server.spotify.client as client_module from spotify_mcp_server.spotify.client import ( SpotifyAPIError, SpotifyClient, _retry_after, sanitize_spotify_payload, ) pytestmark = pytest.mark.anyio class Tokens: def __init__(self) -> None: self.calls: list[bool] = [] async def access_token(self, *, force_refresh: bool = False) -> str: self.calls.append(force_refresh) return "refreshed" if force_refresh else "initial" def test_sanitize_preserves_shapes_and_removes_deprecated_fields() -> None: payload = { "type": "playlist", "tracks": {"items": []}, "items": [ { "type": "track", "name": "Song", "available_markets": ["US"], "external_ids": {"isrc": "x"}, "album": {"type": "album", "label": "label", "name": "Album"}, } ], } assert sanitize_spotify_payload(payload) == { "type": "playlist", "items": [ { "type": "track", "name": "Song", "external_ids": {"isrc": "x"}, "album": {"type": "album", "name": "Album"}, } ], } assert sanitize_spotify_payload({"type": "user", "followers": {"total": 3}}) == { "type": "user", "followers": {"total": 3}, } async def test_client_blocks_deprecated_or_unapproved_operation() -> None: async with httpx2.AsyncClient( transport=httpx2.MockTransport(lambda _: httpx2.Response(204)) ) as http: client = SpotifyClient(Tokens(), client=http) with pytest.raises(ValueError, match="not in the audited allowlist"): await client.request("PUT", "/me/tracks", params={"ids": "legacy"}) async def test_request_refresh_retry_and_no_content() -> None: seen = 0 async def handler(request: httpx2.Request) -> httpx2.Response: nonlocal seen seen += 1 if seen == 1: return httpx2.Response(401, json={"error": {"message": "expired"}}) assert request.headers["Authorization"] == "Bearer refreshed" return httpx2.Response(204) tokens = Tokens() transport = httpx2.MockTransport(handler) async with httpx2.AsyncClient(transport=transport) as http: client = SpotifyClient(tokens, client=http, max_retries=0) assert await client.request("PUT", "/me/player/pause") is None assert tokens.calls == [False, True] async def test_request_treats_non_json_success_body_as_no_content() -> None: """Shuffle and repeat answer 200 with a short non-JSON body instead of 204.""" async def handler(_: httpx2.Request) -> httpx2.Response: return httpx2.Response(200, content=b"OK") async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: client = SpotifyClient(Tokens(), client=http, max_retries=0) assert await client.request("PUT", "/me/player/shuffle", params={"state": True}) is None assert await client.request("PUT", "/me/player/repeat", params={"state": "off"}) is None async def test_request_retries_rate_limit_then_returns_payload() -> None: sleeps: list[float] = [] seen = 0 async def handler(_: httpx2.Request) -> httpx2.Response: nonlocal seen seen += 1 if seen == 1: return httpx2.Response(429, headers={"Retry-After": "0"}, json={"error": {}}) return httpx2.Response(200, json={"type": "track", "name": "ok", "popularity": 100}) async def sleep(value: float) -> None: sleeps.append(value) async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: client = SpotifyClient(Tokens(), client=http, max_retries=1, sleep=sleep) assert await client.request("GET", "/tracks/x") == {"type": "track", "name": "ok"} assert sleeps == [0.0] async def test_request_raises_structured_error() -> None: async def handler(_: httpx2.Request) -> httpx2.Response: return httpx2.Response( 429, headers={"Retry-After": "nonsense"}, json={"error": {"message": "slow", "reason": "QUOTA_EXCEEDED"}}, ) async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: client = SpotifyClient(Tokens(), client=http, max_retries=0) with pytest.raises(SpotifyAPIError) as raised: await client.request("GET", "/search") assert raised.value.reason == "QUOTA_EXCEEDED" assert raised.value.retry_after_seconds is None def test_retry_after_parses_http_date(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(client_module.time, "time", lambda: 1_000.0) response = httpx2.Response(429, headers={"Retry-After": "Thu, 01 Jan 1970 00:17:10 GMT"}) assert _retry_after(response) == 30.0 async def test_request_retries_network_error_then_structures_failure() -> None: attempts = 0 sleeps: list[float] = [] async def handler(request: httpx2.Request) -> httpx2.Response: nonlocal attempts attempts += 1 raise httpx2.ConnectError("offline", request=request) async def sleep(value: float) -> None: sleeps.append(value) async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: client = SpotifyClient(Tokens(), client=http, max_retries=1, sleep=sleep) with pytest.raises(SpotifyAPIError, match="network request failed") as raised: await client.request("GET", "/tracks/x") assert raised.value.status_code == 0 assert attempts == 2 assert sleeps == [0.5] async def test_paged_combines_items_and_honors_budget() -> None: async def handler(request: httpx2.Request) -> httpx2.Response: offset = request.url.params.get("offset", "0") if offset == "0": return httpx2.Response( 200, json={ "items": [{"id": "one"}], "next": "https://api.spotify.com/v1/me/tracks?offset=1", "total": 3, }, ) return httpx2.Response( 200, json={ "items": [{"id": "two"}], "next": "https://api.spotify.com/v1/me/tracks?offset=2", "total": 3, }, ) async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: client = SpotifyClient(Tokens(), client=http) page = await client.paged("/me/tracks", params={"offset": 0}, max_pages=2) assert [item["id"] for item in page["items"]] == ["one", "two"] assert page["pages_fetched"] == 2 assert page["next"].endswith("offset=2") async def test_paged_combines_nested_search_containers() -> None: async def handler(request: httpx2.Request) -> httpx2.Response: offset = request.url.params.get("offset", "0") return httpx2.Response( 200, json={ "albums": { "items": [{"id": offset}], "next": ( "https://api.spotify.com/v1/search?q=x&type=album&offset=1" if offset == "0" else None ), } }, ) async with httpx2.AsyncClient(transport=httpx2.MockTransport(handler)) as http: client = SpotifyClient(Tokens(), client=http) page = await client.paged( "/search", params={"q": "x", "type": "album", "offset": 0}, max_pages=2, container_key="albums", ) assert [item["id"] for item in page["items"]] == ["0", "1"] assert page["pages_fetched"] == 2 assert page["next"] is None async def test_paged_normalizes_missing_initial_container() -> None: async with httpx2.AsyncClient( transport=httpx2.MockTransport(lambda _: httpx2.Response(200, json={})) ) as http: client = SpotifyClient(Tokens(), client=http) with pytest.raises(SpotifyAPIError, match=r"albums.*paging object"): await client.paged( "/search", params={"q": "x", "type": "album"}, max_pages=1, container_key="albums" ) async def test_paged_normalizes_missing_next_container() -> None: responses = [ httpx2.Response( 200, json={ "albums": { "items": [{"id": "one"}], "next": "https://api.spotify.com/v1/search?offset=1", } }, ), httpx2.Response(200, json={}), ] async with httpx2.AsyncClient( transport=httpx2.MockTransport(lambda _: responses.pop(0)) ) as http: client = SpotifyClient(Tokens(), client=http) with pytest.raises(SpotifyAPIError, match=r"albums.*paging object"): await client.paged( "/search", params={"q": "x", "type": "album"}, max_pages=2, container_key="albums" )