/
/
/
1"""Tests for the RecommendationPayloadMixin cached-payload helper."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7from typing import Any, cast
8from unittest.mock import AsyncMock, Mock
9
10import pytest
11from music_assistant_models.enums import ImageType, MediaType, RecommendationFolderType
12from music_assistant_models.media_items import ItemMapping, MediaItemImage, RecommendationFolder
13from music_assistant_models.unique_list import UniqueList
14
15from music_assistant.models.recommendation_payload import (
16 _PAYLOAD_CACHE_KEY,
17 RecommendationPayloadMixin,
18)
19from tests.common import collect_loop_errors
20
21INSTANCE_ID = "test_payload--instance1"
22
23
24def _payload_dicts(payload: list[RecommendationFolder]) -> list[dict[str, Any]]:
25 """
26 Serialize folders to dicts for equality checks across distinct instances.
27
28 RecommendationFolder inherits an __eq__ that only accepts MediaItem/ItemMapping
29 instances (BrowseFolder, its own base, is neither), so two structurally-identical
30 but distinct RecommendationFolder objects always compare unequal - only the exact
31 same object survives `==`. Comparing to_dict() output instead asserts the fields
32 actually match, which is what a persisted-cache round trip (a fresh reconstruction
33 via from_dict) needs.
34 """
35 return [folder.to_dict() for folder in payload]
36
37
38class _UnloadableBase:
39 """Stands in for the Provider base at the end of the cooperative unload chain."""
40
41 unload_chain_called = False
42
43 async def unload(self, is_removed: bool = False) -> None:
44 self.unload_chain_called = True
45
46
47class _PayloadProvider(RecommendationPayloadMixin, _UnloadableBase):
48 """Minimal host implementing the mixin's requirements, with a dict-backed fake cache."""
49
50 # the mixin is typed against the real Provider base; the fake overrides its
51 # (final) identity properties with plain values, which is fine at runtime
52 domain = "test_payload" # type: ignore[misc]
53 instance_id = INSTANCE_ID # type: ignore[misc]
54
55 def __init__(self, fetch: AsyncMock) -> None:
56 self.logger = logging.getLogger(__name__)
57 self._fetch = fetch
58 self._cache_store: dict[str, Any] = {}
59 # freshness the fake cache reports for a hit; set False to simulate an
60 # expired persistent entry (returned as stale data, is_fresh=False)
61 self.cache_is_fresh = True
62 self.background_tasks: list[asyncio.Future[Any]] = []
63 self.mass = Mock()
64 self.mass.cache.get_with_freshness = AsyncMock(side_effect=self._cache_get)
65 self.mass.cache.set = AsyncMock(side_effect=self._cache_set)
66 self.mass.create_task = Mock(side_effect=self._create_task)
67
68 async def _fetch_recommendation_payload(self) -> list[RecommendationFolder]:
69 result: list[RecommendationFolder] = await self._fetch()
70 return result
71
72 def seed_cache(self, payload: list[RecommendationFolder]) -> None:
73 """Pre-fill the persistent cache store, as a previous run's store task would."""
74 self._cache_store[_PAYLOAD_CACHE_KEY] = _payload_dicts(payload)
75
76 async def _cache_get(self, key: str, **kwargs: Any) -> tuple[Any, bool, bool]:
77 if key not in self._cache_store:
78 return None, False, False
79 data = self._cache_store[key]
80 base_class = kwargs.get("base_class")
81 if base_class is not None and data is not None:
82 # mirrors production get_with_freshness: a list of dicts reconstructs as a
83 # list of base_class instances, a single dict as one base_class instance
84 if isinstance(data, list):
85 return [base_class.from_dict(item) for item in data], self.cache_is_fresh, True
86 return base_class.from_dict(data), self.cache_is_fresh, True
87 return data, self.cache_is_fresh, True
88
89 async def _cache_set(self, key: str, data: Any, **kwargs: Any) -> None:
90 # mirrors production: entries are stored serialized (to_dict), reconstructed
91 # via base_class.from_dict above on a hit
92 if isinstance(data, list):
93 self._cache_store[key] = [item.to_dict() for item in data]
94 elif data is not None:
95 self._cache_store[key] = data.to_dict()
96 else:
97 self._cache_store[key] = None
98
99 def _create_task(self, target: Any, *args: Any, **kwargs: Any) -> asyncio.Future[Any]:
100 task: asyncio.Future[Any] = asyncio.ensure_future(target)
101 self.background_tasks.append(task)
102 return task
103
104
105async def _drain_background(provider: _PayloadProvider) -> None:
106 """Let all background tasks finish, including tasks they spawn while draining."""
107 while pending := [task for task in provider.background_tasks if not task.done()]:
108 await asyncio.gather(*pending, return_exceptions=True)
109
110
111def _make_payload() -> list[RecommendationFolder]:
112 """Build a two-folder payload with items and all identity/presentation fields set."""
113 return [
114 RecommendationFolder(
115 item_id=f"{INSTANCE_ID}_editorial",
116 provider=INSTANCE_ID,
117 name="Editorial",
118 translation_key="editorial_picks",
119 icon="mdi-star",
120 subtitle="Picked for you",
121 enabled_by_default=False,
122 is_playable=True,
123 media_type=MediaType.PLAYLIST,
124 type=RecommendationFolderType.TIMELINE,
125 image=MediaItemImage(
126 type=ImageType.THUMB,
127 path="http://backend/editorial.jpg",
128 provider=INSTANCE_ID,
129 remotely_accessible=True,
130 ),
131 items=UniqueList(
132 [
133 ItemMapping(
134 media_type=MediaType.TRACK,
135 item_id="t1",
136 provider=INSTANCE_ID,
137 name="Track One",
138 )
139 ]
140 ),
141 ),
142 RecommendationFolder(
143 item_id=f"{INSTANCE_ID}_charts",
144 provider=INSTANCE_ID,
145 name="Charts",
146 items=UniqueList(
147 [
148 ItemMapping(
149 media_type=MediaType.PLAYLIST,
150 item_id="p1",
151 provider=INSTANCE_ID,
152 name="Chart Playlist",
153 )
154 ]
155 ),
156 ),
157 ]
158
159
160def _make_fresh_payload() -> list[RecommendationFolder]:
161 """Build a single-folder payload distinct from _make_payload, for refresh scenarios."""
162 return [
163 RecommendationFolder(
164 item_id=f"{INSTANCE_ID}_fresh",
165 provider=INSTANCE_ID,
166 name="Fresh",
167 items=UniqueList(
168 [
169 ItemMapping(
170 media_type=MediaType.TRACK,
171 item_id="t2",
172 provider=INSTANCE_ID,
173 name="Track Two",
174 )
175 ]
176 ),
177 )
178 ]
179
180
181@pytest.mark.asyncio
182async def test_rows_and_items_share_one_payload_fetch() -> None:
183 """Rows and two items calls are served from a single backend fetch via memory."""
184 payload = _make_payload()
185 fetch = AsyncMock(return_value=payload)
186 provider = _PayloadProvider(fetch)
187
188 rows = await provider._recommendation_rows_from_payload()
189 items_editorial = await provider._recommendation_items_from_payload(f"{INSTANCE_ID}_editorial")
190 items_charts = await provider._recommendation_items_from_payload(f"{INSTANCE_ID}_charts")
191
192 fetch.assert_awaited_once()
193 assert [row.item_id for row in rows] == [
194 f"{INSTANCE_ID}_editorial",
195 f"{INSTANCE_ID}_charts",
196 ]
197 # warm calls serve the payload straight from memory: the very objects the
198 # backend fetch returned, with no cache-db round trip / re-deserialization
199 assert items_editorial is payload[0].items
200 assert items_charts is payload[1].items
201 # the fetch also stored the payload persistently, under the legacy cache key
202 await _drain_background(provider)
203 assert provider._cache_store[_PAYLOAD_CACHE_KEY] == _payload_dicts(payload)
204
205
206@pytest.mark.asyncio
207async def test_cold_start_rows_then_items_before_store_lands_fetches_once() -> None:
208 """The cold-start rows->items sequence does one fetch even if the store hasn't landed."""
209 payload = _make_payload()
210 fetch = AsyncMock(return_value=payload)
211 provider = _PayloadProvider(fetch)
212 # hold back the persistent store so the items call cannot be served by the cache db
213 store_gate = asyncio.Event()
214 plain_set = provider._cache_set
215
216 async def _gated_set(key: str, data: Any, **kwargs: Any) -> None:
217 await store_gate.wait()
218 await plain_set(key, data, **kwargs)
219
220 provider.mass.cache.set = AsyncMock(side_effect=_gated_set) # type: ignore[method-assign]
221
222 rows = await provider._recommendation_rows_from_payload()
223 items = await provider._recommendation_items_from_payload(f"{INSTANCE_ID}_editorial")
224
225 fetch.assert_awaited_once()
226 assert len(rows) == 2
227 assert items is payload[0].items
228
229 store_gate.set()
230 await _drain_background(provider)
231 assert provider._cache_store[_PAYLOAD_CACHE_KEY] == _payload_dicts(payload)
232
233
234@pytest.mark.asyncio
235async def test_single_flight_concurrent_cold_calls_fetch_once() -> None:
236 """N concurrent cold callers share one in-flight backend fetch."""
237 payload = _make_payload()
238 gate = asyncio.Event()
239
240 async def _gated_fetch() -> list[RecommendationFolder]:
241 await gate.wait()
242 return payload
243
244 fetch = AsyncMock(side_effect=_gated_fetch)
245 provider = _PayloadProvider(fetch)
246
247 tasks = [asyncio.create_task(provider._recommendation_payload()) for _ in range(5)]
248 await asyncio.sleep(0)
249 gate.set()
250 results = await asyncio.gather(*tasks)
251
252 assert fetch.await_count == 1
253 assert all(result == payload for result in results)
254
255
256@pytest.mark.asyncio
257async def test_unknown_item_id_returns_empty() -> None:
258 """An item_id not present in the payload yields an empty UniqueList."""
259 fetch = AsyncMock(return_value=_make_payload())
260 provider = _PayloadProvider(fetch)
261
262 result = await provider._recommendation_items_from_payload("bogus_row")
263
264 assert isinstance(result, UniqueList)
265 assert len(result) == 0
266
267
268@pytest.mark.asyncio
269async def test_rows_have_empty_items_but_preserve_all_fields() -> None:
270 """Rows are stripped of items while keeping every other field of the payload folder."""
271 payload = _make_payload()
272 fetch = AsyncMock(return_value=payload)
273 provider = _PayloadProvider(fetch)
274
275 rows = await provider._recommendation_rows_from_payload()
276
277 editorial = rows[0]
278 assert len(editorial.items) == 0
279 assert editorial.item_id == f"{INSTANCE_ID}_editorial"
280 assert editorial.provider == INSTANCE_ID
281 assert editorial.name == "Editorial"
282 assert editorial.translation_key == "editorial_picks"
283 assert editorial.icon == "mdi-star"
284 assert editorial.subtitle == "Picked for you"
285 assert editorial.enabled_by_default is False
286 # presentation/behavior fields survive the copy too (regression: an explicit
287 # field-by-field copy silently dropped these)
288 assert editorial.is_playable is True
289 assert editorial.media_type == MediaType.PLAYLIST
290 assert editorial.type == RecommendationFolderType.TIMELINE
291 assert editorial.image is not None
292 assert editorial.image.path == "http://backend/editorial.jpg"
293 # stripping produced fresh copies: the payload folders keep their items
294 assert editorial is not payload[0]
295 assert len(payload[0].items) == 1
296
297
298@pytest.mark.asyncio
299async def test_stale_persistent_entry_served_with_single_background_refresh() -> None:
300 """A stale persisted payload is served immediately while exactly one refresh runs."""
301 stale = _make_payload()
302 fresh = _make_fresh_payload()
303 gate = asyncio.Event()
304
305 async def _gated_fetch() -> list[RecommendationFolder]:
306 await gate.wait()
307 return fresh
308
309 fetch = AsyncMock(side_effect=_gated_fetch)
310 provider = _PayloadProvider(fetch)
311 provider.seed_cache(stale)
312 provider.cache_is_fresh = False
313
314 # the stale payload is served without waiting for the backend
315 rows = await provider._recommendation_rows_from_payload()
316 assert [row.item_id for row in rows] == [folder.item_id for folder in stale]
317 # exactly one background refresh was scheduled and is now blocked in the backend call
318 await asyncio.sleep(0)
319 assert fetch.await_count == 1
320 # a second call while the refresh runs serves stale data and schedules no duplicate
321 items = await provider._recommendation_items_from_payload(f"{INSTANCE_ID}_editorial")
322 assert [item.item_id for item in items] == ["t1"]
323 await asyncio.sleep(0)
324 assert fetch.await_count == 1
325
326 gate.set()
327 await _drain_background(provider)
328
329 # the refreshed payload replaced both the in-memory payload and the cache entry
330 assert await provider._recommendation_payload() == fresh
331 assert provider._cache_store[_PAYLOAD_CACHE_KEY] == _payload_dicts(fresh)
332 assert fetch.await_count == 1
333
334
335@pytest.mark.asyncio
336async def test_fresh_persistent_entry_serves_cold_instance_without_fetch() -> None:
337 """A fresh persisted payload warms a cold instance without any backend fetch."""
338 payload = _make_payload()
339 fetch = AsyncMock(return_value=payload)
340 provider = _PayloadProvider(fetch)
341 provider.seed_cache(payload)
342
343 rows = await provider._recommendation_rows_from_payload()
344 items = await provider._recommendation_items_from_payload(f"{INSTANCE_ID}_editorial")
345
346 fetch.assert_not_awaited()
347 # the cache db was read exactly once; the items call was served from memory
348 cast("AsyncMock", provider.mass.cache.get_with_freshness).assert_awaited_once()
349 assert [row.item_id for row in rows] == [folder.item_id for folder in payload]
350 assert [item.item_id for item in items] == ["t1"]
351
352
353@pytest.mark.asyncio
354async def test_refresh_fetches_fresh_and_stores() -> None:
355 """_refresh_recommendation_payload bypasses cached data and updates memory + cache."""
356 stale = _make_payload()
357 fetch = AsyncMock(return_value=stale)
358 provider = _PayloadProvider(fetch)
359 # warm memory and the persistent cache with the stale payload
360 await provider._recommendation_payload()
361 await _drain_background(provider)
362 fresh = _make_fresh_payload()
363 fetch.return_value = fresh
364
365 result = await provider._refresh_recommendation_payload()
366
367 assert result == fresh
368 assert fetch.await_count == 2
369 await _drain_background(provider)
370 # the refresh stored under the same (legacy) key, replacing the stale entry
371 assert set(provider._cache_store) == {_PAYLOAD_CACHE_KEY}
372 assert provider._cache_store[_PAYLOAD_CACHE_KEY] == _payload_dicts(fresh)
373 # subsequent payload calls serve the refreshed payload from memory, no new fetch
374 assert await provider._recommendation_payload() == fresh
375 assert fetch.await_count == 2
376
377
378@pytest.mark.asyncio
379async def test_refresh_raising_fetch_does_not_store_and_does_not_poison_retry() -> None:
380 """A raising refresh propagates without storing; cached data still serves, and retries."""
381 stale = _make_payload()
382 fetch = AsyncMock(return_value=stale)
383 provider = _PayloadProvider(fetch)
384 # warm memory and the persistent cache with the stale payload
385 await provider._recommendation_payload()
386 await _drain_background(provider)
387 cast("AsyncMock", provider.mass.cache.set).reset_mock()
388
389 fetch.side_effect = RuntimeError("backend down")
390
391 with pytest.raises(RuntimeError, match="backend down"):
392 await provider._refresh_recommendation_payload()
393
394 cast("AsyncMock", provider.mass.cache.set).assert_not_awaited()
395 # the previously fetched payload still serves from memory
396 assert await provider._recommendation_payload() == stale
397 assert fetch.await_count == 2
398
399 fresh = _make_fresh_payload()
400 fetch.side_effect = None
401 fetch.return_value = fresh
402
403 result = await provider._refresh_recommendation_payload()
404
405 assert result == fresh
406 await _drain_background(provider)
407 cast("AsyncMock", provider.mass.cache.set).assert_awaited_once()
408 assert await provider._recommendation_payload() == fresh
409 assert fetch.await_count == 3
410
411
412@pytest.mark.asyncio
413async def test_refresh_single_flight_concurrent_calls_fetch_once() -> None:
414 """Concurrent refresh callers share one in-flight backend fetch."""
415 payload = _make_payload()
416 gate = asyncio.Event()
417
418 async def _gated_fetch() -> list[RecommendationFolder]:
419 await gate.wait()
420 return payload
421
422 fetch = AsyncMock(side_effect=_gated_fetch)
423 provider = _PayloadProvider(fetch)
424
425 tasks = [asyncio.create_task(provider._refresh_recommendation_payload()) for _ in range(5)]
426 await asyncio.sleep(0)
427 gate.set()
428 results = await asyncio.gather(*tasks)
429
430 assert fetch.await_count == 1
431 assert all(result == payload for result in results)
432
433
434@pytest.mark.asyncio
435async def test_cancelled_waiter_does_not_cancel_shared_fetch() -> None:
436 """
437 A timed-out caller leaves the shared fetch alone: other waiters still get the payload.
438
439 Regression test: the controller's asyncio.timeout cancels the calling task; if that
440 cancellation propagated into the shared single-flight task it would raise
441 CancelledError in every other waiter and prevent the cache from warming.
442 """
443 payload = _make_payload()
444 gate = asyncio.Event()
445
446 async def _gated_fetch() -> list[RecommendationFolder]:
447 await gate.wait()
448 return payload
449
450 fetch = AsyncMock(side_effect=_gated_fetch)
451 provider = _PayloadProvider(fetch)
452
453 fast_caller = asyncio.create_task(provider._recommendation_payload())
454 slow_caller = asyncio.create_task(provider._recommendation_payload())
455 await asyncio.sleep(0)
456 # simulate the rows call timing out while the fetch is still in flight
457 fast_caller.cancel()
458 gate.set()
459
460 assert await slow_caller == payload
461 with pytest.raises(asyncio.CancelledError):
462 await fast_caller
463 # the shared fetch completed exactly once despite the cancelled waiter
464 fetch.assert_awaited_once()
465
466
467@pytest.mark.asyncio
468async def test_sole_cancelled_waiter_fetch_still_warms_memory_and_cache() -> None:
469 """A cold fetch whose only waiter is cancelled still completes and warms memory + cache."""
470 payload = _make_payload()
471 gate = asyncio.Event()
472
473 async def _gated_fetch() -> list[RecommendationFolder]:
474 await gate.wait()
475 return payload
476
477 fetch = AsyncMock(side_effect=_gated_fetch)
478 provider = _PayloadProvider(fetch)
479
480 sole_caller = asyncio.create_task(provider._recommendation_payload())
481 await asyncio.sleep(0)
482 # the rows timeout case: the only waiter is cancelled mid-fetch
483 sole_caller.cancel()
484 with pytest.raises(asyncio.CancelledError):
485 await sole_caller
486
487 gate.set()
488 await _drain_background(provider)
489
490 # the fetch completed anyway: memory and cache are warm, no second fetch
491 assert await provider._recommendation_payload() == payload
492 fetch.assert_awaited_once()
493 assert provider._cache_store[_PAYLOAD_CACHE_KEY] == _payload_dicts(payload)
494
495
496@pytest.mark.asyncio
497async def test_failed_fetch_propagates_to_all_waiters_and_retries() -> None:
498 """A failing fetch raises in every concurrent waiter and does not poison later calls."""
499 payload = _make_payload()
500 gate = asyncio.Event()
501
502 async def _failing_fetch() -> list[RecommendationFolder]:
503 await gate.wait()
504 raise RuntimeError("backend down")
505
506 fetch = AsyncMock(side_effect=_failing_fetch)
507 provider = _PayloadProvider(fetch)
508
509 tasks = [asyncio.create_task(provider._recommendation_payload()) for _ in range(3)]
510 await asyncio.sleep(0)
511 gate.set()
512 results = await asyncio.gather(*tasks, return_exceptions=True)
513
514 assert len(results) == 3
515 assert all(isinstance(result, RuntimeError) for result in results)
516
517 fetch.side_effect = None
518 fetch.return_value = payload
519 assert await provider._recommendation_payload() == payload
520 assert fetch.await_count == 2
521
522
523@pytest.mark.asyncio
524async def test_cancelled_waiter_of_a_failing_fetch_logs_no_loop_error() -> None:
525 """A fetch failing after a caller timed out is not reported to the loop handler."""
526 gate = asyncio.Event()
527
528 async def _failing_fetch() -> list[RecommendationFolder]:
529 await gate.wait()
530 raise RuntimeError("backend down")
531
532 provider = _PayloadProvider(AsyncMock(side_effect=_failing_fetch))
533
534 with collect_loop_errors() as reported:
535 fast_caller = asyncio.create_task(provider._recommendation_payload())
536 slow_caller = asyncio.create_task(provider._recommendation_payload())
537 await asyncio.sleep(0)
538 # simulate the rows call timing out before the fetch fails
539 fast_caller.cancel()
540 with pytest.raises(asyncio.CancelledError):
541 await fast_caller
542 gate.set()
543 with pytest.raises(RuntimeError, match="backend down"):
544 await slow_caller
545
546 assert reported == []
547
548
549@pytest.mark.asyncio
550async def test_unload_cancels_inflight_cold_fetch_and_continues_chain() -> None:
551 """unload() cancels a hanging cold fetch and still runs the provider unload chain."""
552 started = asyncio.Event()
553
554 async def _hanging_fetch() -> list[RecommendationFolder]:
555 started.set()
556 await asyncio.Event().wait()
557 return []
558
559 provider = _PayloadProvider(AsyncMock(side_effect=_hanging_fetch))
560 caller = asyncio.create_task(provider._recommendation_payload())
561 await started.wait()
562 task = provider._recommendation_payload_task
563 assert task is not None
564
565 await provider.unload()
566
567 assert provider.unload_chain_called is True
568 # the fetch was cancelled, awaited and its handle cleared before the chain continued
569 assert task.cancelled()
570 assert provider._recommendation_payload_task is None
571 with pytest.raises(asyncio.CancelledError):
572 await caller
573
574
575@pytest.mark.asyncio
576async def test_unload_cancels_background_refresh() -> None:
577 """unload() cancels an in-flight stale-payload background refresh."""
578 stale = _make_payload()
579 started = asyncio.Event()
580
581 async def _hanging_fetch() -> list[RecommendationFolder]:
582 started.set()
583 await asyncio.Event().wait()
584 return []
585
586 provider = _PayloadProvider(AsyncMock(side_effect=_hanging_fetch))
587 provider.seed_cache(stale)
588 provider.cache_is_fresh = False
589
590 # serves the stale payload and schedules the (hanging) background refresh
591 assert _payload_dicts(await provider._recommendation_payload()) == _payload_dicts(stale)
592 await started.wait()
593 task = provider._recommendation_refresh_task
594 assert task is not None
595
596 await provider.unload()
597
598 assert provider.unload_chain_called is True
599 assert task.cancelled()
600 assert provider._recommendation_refresh_task is None
601
602
603@pytest.mark.asyncio
604async def test_unload_without_inflight_tasks_is_a_noop() -> None:
605 """unload() on an idle provider just continues the unload chain."""
606 provider = _PayloadProvider(AsyncMock(return_value=[]))
607
608 await provider.unload()
609
610 assert provider.unload_chain_called is True
611