/
/
/
1"""Tests for Spotify playlist track pagination."""
2
3import asyncio
4from collections import OrderedDict
5from typing import Any, NamedTuple
6from unittest.mock import AsyncMock, MagicMock, call
7
8import pytest
9from music_assistant_models.errors import MediaNotFoundError
10
11from music_assistant.providers.spotify.provider import (
12 _PLAYLIST_PAGINATION_STATE_LIMIT,
13 SpotifyProvider,
14)
15from tests.common import use_real_create_task
16
17PLAYLIST_ID = "private-playlist"
18
19
20class SpotifyPlaylistHarness(NamedTuple):
21 """Container for a Spotify provider and its mocked API collaborators."""
22
23 provider: SpotifyProvider
24 get_playlist: AsyncMock
25 requires_global: AsyncMock
26 get_metadata: AsyncMock
27 get_page: AsyncMock
28 set_global: AsyncMock
29
30
31def _make_provider(instance_id: str = "spotify--test") -> SpotifyPlaylistHarness:
32 """Return a Spotify provider with isolated playlist API mocks."""
33 provider = object.__new__(SpotifyProvider)
34 provider.config = MagicMock(instance_id=instance_id)
35 provider.manifest = MagicMock(domain="spotify")
36 provider.logger = MagicMock()
37 provider._sp_user = {"id": "test-user", "display_name": "Test User"}
38 provider._playlist_pagination_states = OrderedDict()
39 provider.dev_session_active = True
40
41 mass = MagicMock()
42 mass.cache.get_with_freshness = AsyncMock(return_value=(None, False, False))
43 mass.cache.set = AsyncMock()
44
45 use_real_create_task(mass)
46 provider.mass = mass
47
48 get_playlist = AsyncMock()
49 requires_global = AsyncMock(return_value=False)
50 get_metadata = AsyncMock()
51 get_page = AsyncMock()
52 set_global = AsyncMock()
53 provider.get_playlist = get_playlist # type: ignore[method-assign]
54 provider._playlist_requires_global_token = requires_global # type: ignore[method-assign]
55 provider._get_paginated_meta = get_metadata # type: ignore[method-assign]
56 provider._get_data_with_caching = get_page # type: ignore[method-assign]
57 provider._set_playlist_requires_global_token = set_global # type: ignore[method-assign]
58
59 return SpotifyPlaylistHarness(
60 provider,
61 get_playlist,
62 requires_global,
63 get_metadata,
64 get_page,
65 set_global,
66 )
67
68
69def _spotify_track(track_id: str) -> dict[str, Any]:
70 """Return the minimum Spotify track payload accepted by the parser."""
71 return {
72 "id": track_id,
73 "name": track_id,
74 "duration_ms": 180000,
75 "external_urls": {"spotify": f"https://open.spotify.com/track/{track_id}"},
76 "is_local": False,
77 "is_playable": True,
78 "explicit": False,
79 }
80
81
82def _playlist_page(total: int, track_id: str) -> dict[str, Any]:
83 """Return a Spotify playlist page containing one playable track."""
84 return {"total": total, "items": [{"track": _spotify_track(track_id)}]}
85
86
87async def test_multipage_traversal_reuses_pagination_metadata() -> None:
88 """A cold sequential traversal requests metadata once and each valid page once."""
89 harness = _make_provider()
90 harness.get_metadata.return_value = {"etag": "snapshot-etag", "total": 120}
91
92 async def get_page(
93 _endpoint: str, _cache_checksum: str | None, **kwargs: Any
94 ) -> dict[str, Any]:
95 return _playlist_page(120, f"track-{kwargs['offset']}")
96
97 harness.get_page.side_effect = get_page
98
99 pages = [
100 await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=page) for page in range(4)
101 ]
102
103 harness.get_metadata.assert_awaited_once_with(
104 f"playlists/{PLAYLIST_ID}/items",
105 limit=1,
106 offset=0,
107 use_global_session=False,
108 )
109 assert harness.get_page.await_args_list == [
110 call(
111 f"playlists/{PLAYLIST_ID}/items",
112 "snapshot-etag",
113 limit=50,
114 offset=offset,
115 use_global_session=False,
116 )
117 for offset in (0, 50, 100)
118 ]
119 assert [page[0].position for page in pages[:3]] == [1, 51, 101]
120 assert pages[3] == []
121
122
123async def test_second_full_traversal_reuses_etag_page_cache() -> None:
124 """A repeated full traversal refreshes metadata but reuses every unchanged page."""
125 harness = _make_provider()
126 harness.provider.__dict__.pop("_get_paginated_meta")
127 harness.provider.__dict__.pop("_get_data_with_caching")
128 page_cache: dict[tuple[str, str | None], dict[str, Any]] = {}
129
130 async def cache_get(
131 key: str, *, checksum: str | None = None, **_kwargs: Any
132 ) -> dict[str, Any] | None:
133 return page_cache.get((key, checksum))
134
135 async def cache_set(
136 key: str,
137 data: dict[str, Any],
138 *,
139 checksum: str | None = None,
140 **_kwargs: Any,
141 ) -> None:
142 page_cache[(key, checksum)] = data
143
144 async def get_data(_endpoint: str, **kwargs: Any) -> dict[str, Any]:
145 if kwargs["limit"] == 1:
146 return {"etag": "stable-etag", "total": 120}
147 return _playlist_page(120, f"track-{kwargs['offset']}")
148
149 harness.provider.mass.cache.get = AsyncMock(side_effect=cache_get) # type: ignore[method-assign]
150 harness.provider.mass.cache.set = AsyncMock(side_effect=cache_set) # type: ignore[method-assign]
151 get_data_mock = AsyncMock(side_effect=get_data)
152 harness.provider._get_data = get_data_mock # type: ignore[method-assign]
153 get_playlist_tracks: Any = SpotifyProvider.get_playlist_tracks.__wrapped__ # type: ignore[attr-defined]
154
155 traversals = [
156 [await get_playlist_tracks(harness.provider, PLAYLIST_ID, page=page) for page in range(4)]
157 for _ in range(2)
158 ]
159
160 metadata_calls = [args for args in get_data_mock.await_args_list if args.kwargs["limit"] == 1]
161 page_calls = [args for args in get_data_mock.await_args_list if args.kwargs["limit"] == 50]
162 assert len(metadata_calls) == 2
163 assert [args.kwargs["offset"] for args in page_calls] == [0, 50, 100]
164 assert harness.provider.mass.cache.get.await_count == 6
165 assert harness.provider.mass.cache.set.await_count == 3
166 assert [[page[0].position for page in traversal[:3]] for traversal in traversals] == [
167 [1, 51, 101],
168 [1, 51, 101],
169 ]
170 assert traversals[0][3] == traversals[1][3] == []
171
172
173async def test_concurrent_cold_pages_share_inflight_pagination_metadata() -> None:
174 """Concurrent cache refreshes share one in-flight metadata request."""
175 harness = _make_provider()
176 metadata_started = asyncio.Event()
177 release_metadata = asyncio.Event()
178
179 async def get_metadata(*_args: Any, **_kwargs: Any) -> dict[str, Any]:
180 metadata_started.set()
181 await release_metadata.wait()
182 return {"etag": "shared-etag", "total": 100}
183
184 harness.get_metadata.side_effect = get_metadata
185 harness.get_page.side_effect = [
186 _playlist_page(100, "page-one-track"),
187 _playlist_page(100, "page-zero-track"),
188 ]
189
190 page_one_task = asyncio.create_task(harness.provider.get_playlist_tracks(PLAYLIST_ID, page=1))
191 await metadata_started.wait()
192 page_zero_task = asyncio.create_task(harness.provider.get_playlist_tracks(PLAYLIST_ID, page=0))
193 await asyncio.sleep(0)
194 release_metadata.set()
195 await asyncio.gather(page_one_task, page_zero_task)
196
197 harness.get_metadata.assert_awaited_once()
198 assert [args.args[1] for args in harness.get_page.await_args_list] == [
199 "shared-etag",
200 "shared-etag",
201 ]
202
203
204async def test_distinct_playlists_use_independent_pagination_states() -> None:
205 """Different playlists neither block nor replace each other's metadata."""
206 harness = _make_provider()
207 first_metadata_started = asyncio.Event()
208 release_first_metadata = asyncio.Event()
209
210 async def get_metadata(endpoint: str, **_kwargs: Any) -> dict[str, Any]:
211 if endpoint == "playlists/private-a/items":
212 first_metadata_started.set()
213 await release_first_metadata.wait()
214 return {"etag": "private-a-etag", "total": 100}
215 return {"etag": "private-b-etag", "total": 100}
216
217 async def get_page(_endpoint: str, cache_checksum: str | None, **kwargs: Any) -> dict[str, Any]:
218 return _playlist_page(100, f"{cache_checksum}-{kwargs['offset']}")
219
220 harness.get_metadata.side_effect = get_metadata
221 harness.get_page.side_effect = get_page
222
223 first_task = asyncio.create_task(harness.provider.get_playlist_tracks("private-a", page=0))
224 await first_metadata_started.wait()
225 try:
226 await asyncio.wait_for(
227 harness.provider.get_playlist_tracks("private-b", page=0),
228 timeout=1,
229 )
230 finally:
231 release_first_metadata.set()
232 await first_task
233 await harness.provider.get_playlist_tracks("private-a", page=1)
234 await harness.provider.get_playlist_tracks("private-b", page=1)
235
236 assert harness.get_metadata.await_args_list == [
237 call(
238 "playlists/private-a/items",
239 limit=1,
240 offset=0,
241 use_global_session=False,
242 ),
243 call(
244 "playlists/private-b/items",
245 limit=1,
246 offset=0,
247 use_global_session=False,
248 ),
249 ]
250 assert [args.args[1] for args in harness.get_page.await_args_list] == [
251 "private-b-etag",
252 "private-a-etag",
253 "private-a-etag",
254 "private-b-etag",
255 ]
256
257
258async def test_page_zero_refresh_replaces_pagination_snapshot() -> None:
259 """A new page-zero request replaces both the ETag and total for later pages."""
260 harness = _make_provider()
261 harness.get_metadata.side_effect = [
262 {"etag": "old-etag", "total": 120},
263 {"etag": "new-etag", "total": 50},
264 ]
265 harness.get_page.side_effect = [
266 _playlist_page(120, "old-track"),
267 _playlist_page(50, "new-track"),
268 ]
269
270 await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=0)
271 await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=0)
272 guarded_page = await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=1)
273
274 assert harness.get_metadata.await_count == 2
275 assert [args.args[1] for args in harness.get_page.await_args_list] == [
276 "old-etag",
277 "new-etag",
278 ]
279 assert guarded_page == []
280 assert harness.get_page.await_count == 2
281
282
283async def test_cold_nonzero_page_fetches_pagination_metadata() -> None:
284 """A direct nonzero page request fetches metadata before requesting its page."""
285 harness = _make_provider()
286 harness.get_metadata.return_value = {"etag": "cold-etag", "total": 151}
287 harness.get_page.return_value = _playlist_page(151, "page-two-track")
288
289 tracks = await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=2)
290
291 harness.get_metadata.assert_awaited_once()
292 harness.get_page.assert_awaited_once_with(
293 f"playlists/{PLAYLIST_ID}/items",
294 "cold-etag",
295 limit=50,
296 offset=100,
297 use_global_session=False,
298 )
299 assert tracks[0].position == 101
300
301
302async def test_playlist_identities_do_not_share_pagination_metadata() -> None:
303 """Private playlists and liked songs each use their own pagination snapshot."""
304 harness = _make_provider()
305 liked_songs_id = harness.provider._get_liked_songs_playlist_id()
306 harness.get_metadata.side_effect = [
307 {"etag": "private-a-etag", "total": 100},
308 {"etag": "private-b-etag", "total": 100},
309 {"etag": "liked-etag", "total": 100},
310 ]
311 harness.get_page.side_effect = [
312 _playlist_page(100, "private-a-track"),
313 _playlist_page(100, "private-b-track"),
314 _playlist_page(100, "liked-track"),
315 ]
316
317 await harness.provider.get_playlist_tracks("private-a", page=1)
318 await harness.provider.get_playlist_tracks("private-b", page=1)
319 await harness.provider.get_playlist_tracks(liked_songs_id, page=1)
320
321 assert harness.get_metadata.await_args_list == [
322 call(
323 "playlists/private-a/items",
324 limit=1,
325 offset=0,
326 use_global_session=False,
327 ),
328 call(
329 "playlists/private-b/items",
330 limit=1,
331 offset=0,
332 use_global_session=False,
333 ),
334 call("me/tracks", limit=1, offset=0, use_global_session=True),
335 ]
336 assert [args.args[1] for args in harness.get_page.await_args_list] == [
337 "private-a-etag",
338 "private-b-etag",
339 "liked-etag",
340 ]
341 assert harness.get_playlist.await_count == 2
342 assert harness.requires_global.await_count == 2
343
344
345async def test_session_paths_do_not_share_pagination_metadata() -> None:
346 """The same playlist refetches metadata when its required token path changes."""
347 harness = _make_provider()
348 harness.requires_global.side_effect = [False, True]
349 harness.get_metadata.side_effect = [
350 {"etag": "dev-etag", "total": 100},
351 {"etag": "global-etag", "total": 100},
352 ]
353 harness.get_page.side_effect = [
354 _playlist_page(100, "dev-track"),
355 _playlist_page(100, "global-track"),
356 ]
357
358 await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=1)
359 await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=1)
360
361 assert harness.get_metadata.await_args_list == [
362 call(
363 f"playlists/{PLAYLIST_ID}/items",
364 limit=1,
365 offset=0,
366 use_global_session=False,
367 ),
368 call(
369 f"playlists/{PLAYLIST_ID}/items",
370 limit=1,
371 offset=0,
372 use_global_session=True,
373 ),
374 ]
375 assert [
376 (args.args[1], args.kwargs["use_global_session"])
377 for args in harness.get_page.await_args_list
378 ] == [("dev-etag", False), ("global-etag", True)]
379
380
381async def test_restricted_playlist_items_metadata_retries_with_global_session() -> None:
382 """Playlist items hidden from the developer session remain available globally."""
383 harness = _make_provider()
384 harness.get_metadata.side_effect = [
385 MediaNotFoundError("developer session forbidden"),
386 {"etag": "global-etag", "total": 1},
387 ]
388 harness.get_page.return_value = _playlist_page(1, "global-track")
389
390 tracks = await harness.provider.get_playlist_tracks(PLAYLIST_ID)
391
392 assert [args.kwargs["use_global_session"] for args in harness.get_metadata.await_args_list] == [
393 False,
394 True,
395 ]
396 harness.get_page.assert_awaited_once_with(
397 f"playlists/{PLAYLIST_ID}/items",
398 "global-etag",
399 limit=50,
400 offset=0,
401 use_global_session=True,
402 )
403 harness.set_global.assert_awaited_once_with(PLAYLIST_ID)
404 assert tracks[0].item_id == "global-track"
405
406
407async def test_restricted_playlist_page_retries_with_global_session() -> None:
408 """A rejected developer-session page retries the complete request through the global session."""
409 harness = _make_provider()
410 harness.get_metadata.side_effect = [
411 {"etag": "dev-etag", "total": 1},
412 {"etag": "global-etag", "total": 1},
413 ]
414 harness.get_page.side_effect = [
415 MediaNotFoundError("developer session forbidden"),
416 _playlist_page(1, "global-track"),
417 ]
418
419 tracks = await harness.provider.get_playlist_tracks(PLAYLIST_ID)
420
421 assert [
422 (args.args[1], args.kwargs["use_global_session"])
423 for args in harness.get_page.await_args_list
424 ] == [("dev-etag", False), ("global-etag", True)]
425 harness.set_global.assert_awaited_once_with(PLAYLIST_ID)
426 assert tracks[0].item_id == "global-track"
427
428
429@pytest.mark.parametrize(
430 ("dev_session_active", "requires_global", "use_global_session"),
431 [(False, False, False), (True, True, True)],
432)
433async def test_playlist_failure_without_available_fallback_propagates(
434 dev_session_active: bool,
435 requires_global: bool,
436 use_global_session: bool,
437) -> None:
438 """Playlist failures propagate when another Spotify session cannot be tried."""
439 harness = _make_provider()
440 harness.provider.dev_session_active = dev_session_active
441 harness.requires_global.return_value = requires_global
442 harness.get_metadata.side_effect = MediaNotFoundError("playlist unavailable")
443
444 with pytest.raises(MediaNotFoundError):
445 await harness.provider.get_playlist_tracks(PLAYLIST_ID)
446
447 harness.get_metadata.assert_awaited_once_with(
448 f"playlists/{PLAYLIST_ID}/items",
449 limit=1,
450 offset=0,
451 use_global_session=use_global_session,
452 )
453 harness.set_global.assert_not_awaited()
454
455
456async def test_failed_global_playlist_fallback_is_not_cached() -> None:
457 """An unavailable playlist is not marked as requiring the global session."""
458 harness = _make_provider()
459 harness.get_metadata.side_effect = [
460 MediaNotFoundError("developer session forbidden"),
461 MediaNotFoundError("playlist unavailable"),
462 ]
463
464 with pytest.raises(MediaNotFoundError):
465 await harness.provider.get_playlist_tracks(PLAYLIST_ID)
466
467 assert [args.kwargs["use_global_session"] for args in harness.get_metadata.await_args_list] == [
468 False,
469 True,
470 ]
471 harness.set_global.assert_not_awaited()
472
473
474async def test_provider_instances_do_not_share_pagination_metadata() -> None:
475 """Each Spotify provider instance maintains an independent pagination snapshot."""
476 first = _make_provider("spotify--first")
477 second = _make_provider("spotify--second")
478 first.get_metadata.return_value = {"etag": "first-etag", "total": 100}
479 second.get_metadata.return_value = {"etag": "second-etag", "total": 100}
480 first.get_page.return_value = _playlist_page(100, "first-track")
481 second.get_page.return_value = _playlist_page(100, "second-track")
482
483 await first.provider.get_playlist_tracks(PLAYLIST_ID, page=1)
484 await second.provider.get_playlist_tracks(PLAYLIST_ID, page=1)
485
486 first.get_metadata.assert_awaited_once()
487 second.get_metadata.assert_awaited_once()
488 assert first.get_page.await_args is not None
489 assert second.get_page.await_args is not None
490 assert first.get_page.await_args.args[1] == "first-etag"
491 assert second.get_page.await_args.args[1] == "second-etag"
492
493
494async def test_offset_guard_skips_invalid_playlist_page_request() -> None:
495 """Known totals prevent Spotify requests at or beyond the playlist end."""
496 harness = _make_provider()
497 harness.get_metadata.return_value = {"etag": "guard-etag", "total": 50}
498
499 tracks = await harness.provider.get_playlist_tracks(PLAYLIST_ID, page=1)
500
501 harness.get_metadata.assert_awaited_once()
502 harness.get_page.assert_not_awaited()
503 assert tracks == []
504
505
506async def test_playlist_pagination_state_is_bounded() -> None:
507 """Pagination state retains only the most recently accessed playlists."""
508 harness = _make_provider()
509 harness.get_metadata.return_value = {"etag": "guard-etag", "total": 50}
510
511 for index in range(_PLAYLIST_PAGINATION_STATE_LIMIT + 1):
512 await harness.provider.get_playlist_tracks(f"playlist-{index}", page=1)
513
514 assert len(harness.provider._playlist_pagination_states) == _PLAYLIST_PAGINATION_STATE_LIMIT
515