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