Spaces:
Running
Running
File size: 9,127 Bytes
93a887a 0645c76 93a887a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 | 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"
)
|