/
/
/
1"""Tests for the @use_cache decorator."""
2
3import asyncio
4import logging
5from collections.abc import Awaitable, Callable
6from dataclasses import dataclass
7from typing import Any
8from unittest.mock import AsyncMock, patch
9
10import pytest
11from mashumaro import DataClassDictMixin
12from music_assistant_models.errors import MediaNotFoundError
13
14from music_assistant.constants import DB_TABLE_CACHE
15from music_assistant.controllers.cache import CacheController
16from music_assistant.controllers.cache.constants import BYPASS_CACHE
17from music_assistant.controllers.cache.helpers import use_cache
18from music_assistant.mass import MusicAssistant
19
20_PROVIDER = "test_cache_helpers"
21
22
23@dataclass
24class _Payload(DataClassDictMixin):
25 """Minimal model to exercise base_class reconstruction."""
26
27 item_id: str
28 played: bool = False
29
30
31class _Uncopyable:
32 """Object that refuses to be copied."""
33
34 def __deepcopy__(self, memo: dict[int, Any]) -> _Uncopyable:
35 """Raise to simulate a result holding something uncopyable."""
36 raise TypeError("cannot copy this")
37
38
39class _FakeProvider:
40 """Minimal object that satisfies the @use_cache protocol."""
41
42 domain = _PROVIDER
43
44 def __init__(self, mass: MusicAssistant) -> None:
45 self.mass = mass
46 self.calls = 0
47 self.result: str | None = None
48 self.error: Exception | None = None
49 # released by default, so tests that do not gate a call are unaffected
50 self.gate = asyncio.Event()
51 self.gate.set()
52
53 @use_cache(3600)
54 async def fetch(self, item_id: str) -> str | None:
55 """Return the preset result, counting invocations."""
56 return await self._result()
57
58 @use_cache(3600, cache_none=False)
59 async def fetch_no_none(self, item_id: str) -> str | None:
60 """Return the preset result, counting invocations."""
61 return await self._result()
62
63 @use_cache(3600, allow_expired_cache=True)
64 async def fetch_swr(self, item_id: str) -> str | None:
65 """Return the preset result, counting invocations."""
66 return await self._result()
67
68 @use_cache(3600)
69 async def fetch_items(self, item_id: str) -> list[dict[str, Any]]:
70 """Return a freshly built mutable payload, counting invocations."""
71 self.calls += 1
72 await self.gate.wait()
73 return [{"id": item_id, "played": False}]
74
75 @use_cache(3600, base_class=_Payload)
76 async def fetch_models(self, item_id: str) -> list[_Payload]:
77 """Return a freshly built list of models, counting invocations."""
78 self.calls += 1
79 await self.gate.wait()
80 return [_Payload(item_id=item_id)]
81
82 @use_cache(3600)
83 async def fetch_reentrant(self, item_id: str) -> str | None:
84 """Call back into itself once for the same item, counting invocations."""
85 self.calls += 1
86 if self.calls == 1:
87 return await self.fetch_reentrant(item_id)
88 return self.result
89
90 @use_cache(3600)
91 async def fetch_tuples(self, item_id: str) -> list[tuple[str, str, str | None]]:
92 """Return a payload whose shape a serialization round-trip would not survive."""
93 self.calls += 1
94 await self.gate.wait()
95 return [(item_id, "name", None)]
96
97 @use_cache(3600)
98 async def fetch_uncopyable(self, item_id: str) -> _Uncopyable:
99 """Return a payload that cannot be copied, counting invocations."""
100 self.calls += 1
101 await self.gate.wait()
102 return _Uncopyable()
103
104 async def _result(self) -> str | None:
105 """Return the preset result once the gate is open, raising a preset error."""
106 self.calls += 1
107 await self.gate.wait()
108 if self.error is not None:
109 raise self.error
110 return self.result
111
112
113@pytest.fixture
114def provider(cache_controller: CacheController) -> _FakeProvider:
115 """Return a fake provider with @use_cache decorated methods."""
116 return _FakeProvider(cache_controller.mass)
117
118
119async def _get_row(cache_controller: CacheController, key: str) -> Any:
120 """Return the raw cache db row for the given key."""
121 assert cache_controller.database is not None
122 return await cache_controller.database.get_row(
123 DB_TABLE_CACHE, {"category": 0, "provider": _PROVIDER, "key": key}
124 )
125
126
127async def _wait_for_stored(cache_controller: CacheController, key: str) -> None:
128 """Wait until the background store task has written the cache row."""
129 for _ in range(200):
130 if await _get_row(cache_controller, key):
131 return
132 await asyncio.sleep(0.01)
133 pytest.fail(f"cache row for {key} was never written")
134
135
136async def _wait_for(condition: Callable[[], Awaitable[bool]]) -> None:
137 """Wait until the given (async) condition callable returns True."""
138 for _ in range(200):
139 if await condition():
140 return
141 await asyncio.sleep(0.01)
142 pytest.fail("condition was never met")
143
144
145async def _wait_for_flight(provider: _FakeProvider) -> None:
146 """Wait until a gated call is running and every concurrent caller has joined it."""
147
148 async def _started() -> bool:
149 return provider.calls > 0
150
151 await _wait_for(_started)
152 # let the remaining callers reach the in-progress fetch; a caller arriving too late
153 # would start a second one, which every call-count assertion below catches
154 await asyncio.sleep(0.05)
155
156
157# --- result caching (including None results) ---
158
159
160async def test_result_served_from_cache(
161 cache_controller: CacheController, provider: _FakeProvider
162) -> None:
163 """Test that a second call is served from cache without re-invoking the function."""
164 provider.result = "value"
165 assert await provider.fetch("a") == "value"
166 assert provider.calls == 1
167 await _wait_for_stored(cache_controller, "fetch.a")
168 provider.result = "changed"
169 assert await provider.fetch("a") == "value"
170 assert provider.calls == 1
171
172
173async def test_none_result_cached_and_served(
174 cache_controller: CacheController, provider: _FakeProvider
175) -> None:
176 """Test that a None result is cached and served without re-invoking the function."""
177 provider.result = None
178 assert await provider.fetch("a") is None
179 assert provider.calls == 1
180 await _wait_for_stored(cache_controller, "fetch.a")
181 assert await provider.fetch("a") is None
182 assert provider.calls == 1
183
184
185async def test_cache_none_false_retries_and_skips_store(
186 cache_controller: CacheController, provider: _FakeProvider
187) -> None:
188 """Test that cache_none=False re-invokes on None and does not store the None result."""
189 provider.result = None
190 assert await provider.fetch_no_none("a") is None
191 assert provider.calls == 1
192 # give a (wrongly created) store task time to run, then verify nothing was written
193 await asyncio.sleep(0.05)
194 assert await _get_row(cache_controller, "fetch_no_none.a") is None
195 assert await provider.fetch_no_none("a") is None
196 assert provider.calls == 2
197 # once the function returns a real value, it is cached again
198 provider.result = "found"
199 assert await provider.fetch_no_none("a") == "found"
200 assert provider.calls == 3
201 await _wait_for_stored(cache_controller, "fetch_no_none.a")
202 assert await provider.fetch_no_none("a") == "found"
203 assert provider.calls == 3
204
205
206async def test_cache_none_false_ignores_stored_none(
207 cache_controller: CacheController, provider: _FakeProvider
208) -> None:
209 """Test that cache_none=False treats a previously stored None row as a cache miss."""
210 await cache_controller.set("fetch_no_none.a", None, provider=_PROVIDER)
211 provider.result = "fresh"
212 assert await provider.fetch_no_none("a") == "fresh"
213 assert provider.calls == 1
214
215
216# --- stale-while-revalidate ---
217
218
219async def test_swr_serves_stale_and_refreshes(
220 cache_controller: CacheController, provider: _FakeProvider
221) -> None:
222 """Test that an expired entry is served immediately and refreshed in the background."""
223 await cache_controller.set(
224 "fetch_swr.a", "stale", provider=_PROVIDER, expiration=-1, allow_expired_cache=True
225 )
226 provider.result = "fresh"
227 assert await provider.fetch_swr("a") == "stale"
228
229 async def _refreshed() -> bool:
230 return bool(await cache_controller.get("fetch_swr.a", provider=_PROVIDER) == "fresh")
231
232 await _wait_for(_refreshed)
233 assert provider.calls == 1
234 assert await provider.fetch_swr("a") == "fresh"
235 assert provider.calls == 1
236
237
238async def test_wrapper_performs_single_row_fetch(
239 cache_controller: CacheController, provider: _FakeProvider
240) -> None:
241 """Test that one wrapper call does exactly one db row fetch (miss, fresh and stale)."""
242 assert cache_controller.database is not None
243 # cache miss
244 provider.result = "value"
245 with patch.object(
246 cache_controller.database, "get_row", wraps=cache_controller.database.get_row
247 ) as spy:
248 assert await provider.fetch_swr("a") == "value"
249 assert spy.await_count == 1
250 # fresh hit
251 await _wait_for_stored(cache_controller, "fetch_swr.a")
252 with patch.object(
253 cache_controller.database, "get_row", wraps=cache_controller.database.get_row
254 ) as spy:
255 assert await provider.fetch_swr("a") == "value"
256 assert spy.await_count == 1
257 # stale hit (previously fetched the same row twice)
258 await cache_controller.set(
259 "fetch_swr.b", "stale", provider=_PROVIDER, expiration=-1, allow_expired_cache=True
260 )
261 with patch.object(
262 cache_controller.database, "get_row", wraps=cache_controller.database.get_row
263 ) as spy:
264 assert await provider.fetch_swr("b") == "stale"
265 assert spy.await_count == 1
266
267
268# --- single flight ---
269
270
271async def test_concurrent_misses_share_one_fetch(
272 cache_controller: CacheController, provider: _FakeProvider
273) -> None:
274 """Test that concurrent callers on the same key trigger one fetch and one store."""
275 provider.result = "value"
276 provider.gate.clear()
277 with patch.object(cache_controller, "set", wraps=cache_controller.set) as store:
278 tasks = [asyncio.create_task(provider.fetch("a")) for _ in range(3)]
279 await _wait_for_flight(provider)
280 provider.gate.set()
281 assert await asyncio.gather(*tasks) == ["value", "value", "value"]
282 assert provider.calls == 1
283 await _wait_for_stored(cache_controller, "fetch.a")
284 assert store.await_count == 1
285
286
287async def test_each_caller_gets_its_own_result_object(provider: _FakeProvider) -> None:
288 """Test that callers sharing one fetch do not share the result objects."""
289 provider.gate.clear()
290 tasks = [asyncio.create_task(provider.fetch_items("a")) for _ in range(3)]
291 await _wait_for_flight(provider)
292 provider.gate.set()
293 results = await asyncio.gather(*tasks)
294 assert provider.calls == 1
295 assert all(result == [{"id": "a", "played": False}] for result in results)
296 assert len({id(result) for result in results}) == 3
297 assert len({id(result[0]) for result in results}) == 3
298 # mutating one caller's payload, as the podcast resume-state code does, must not
299 # be visible to the other callers
300 results[0][0]["played"] = True
301 assert results[1][0]["played"] is False
302 assert results[2][0]["played"] is False
303
304
305async def test_model_results_are_copied_per_caller(provider: _FakeProvider) -> None:
306 """Test that model results handed to callers sharing a fetch are copies."""
307 provider.gate.clear()
308 tasks = [asyncio.create_task(provider.fetch_models("a")) for _ in range(3)]
309 await _wait_for_flight(provider)
310 provider.gate.set()
311 results = await asyncio.gather(*tasks)
312 assert provider.calls == 1
313 assert all(result == [_Payload(item_id="a")] for result in results)
314 assert len({id(result[0]) for result in results}) == 3
315 results[0][0].played = True
316 assert results[1][0].played is False
317 assert results[2][0].played is False
318
319
320async def test_caller_mutation_does_not_reach_the_other_callers(
321 cache_controller: CacheController, provider: _FakeProvider
322) -> None:
323 """Test that a caller mutating its result right away cannot affect the others."""
324
325 async def _fetch_and_mutate() -> list[dict[str, Any]]:
326 result = await provider.fetch_items("a")
327 # runs before the other callers resume, so a copy taken later would pick this up
328 result[0]["played"] = True
329 return result
330
331 provider.gate.clear()
332 tasks = [
333 asyncio.create_task(_fetch_and_mutate()),
334 asyncio.create_task(provider.fetch_items("a")),
335 asyncio.create_task(provider.fetch_items("a")),
336 ]
337 await _wait_for_flight(provider)
338 provider.gate.set()
339 mutated, *others = await asyncio.gather(*tasks)
340 assert provider.calls == 1
341 assert mutated[0]["played"] is True
342 assert [result[0]["played"] for result in others] == [False, False]
343 # the entry is written from the fetched objects, which no caller was handed
344 await _wait_for_stored(cache_controller, "fetch_items.a")
345 assert await cache_controller.get("fetch_items.a", provider=_PROVIDER) == [
346 {"id": "a", "played": False}
347 ]
348
349
350async def test_reentrant_call_on_the_same_key_completes(provider: _FakeProvider) -> None:
351 """Test that a body calling back into itself for its own key does not wait on itself."""
352 provider.result = "value"
353 async with asyncio.timeout(5):
354 assert await provider.fetch_reentrant("a") == "value"
355 assert provider.calls == 2
356
357
358async def test_result_shape_survives_being_copied(provider: _FakeProvider) -> None:
359 """Test that callers sharing a fetch get the result shape the function returned."""
360 provider.gate.clear()
361 tasks = [asyncio.create_task(provider.fetch_tuples("a")) for _ in range(3)]
362 await _wait_for_flight(provider)
363 provider.gate.set()
364 results = await asyncio.gather(*tasks)
365 assert provider.calls == 1
366 assert all(result == [("a", "name", None)] for result in results)
367
368
369async def test_tuple_result_shape_survives_a_cache_hit(
370 cache_controller: CacheController, provider: _FakeProvider
371) -> None:
372 """Test that a cached tuple is rebuilt with all of its members, including the None ones."""
373 assert await provider.fetch_tuples("a") == [("a", "name", None)]
374 await _wait_for_stored(cache_controller, "fetch_tuples.a")
375 assert await provider.fetch_tuples("a") == [("a", "name", None)]
376 assert provider.calls == 1
377
378
379async def test_uncopyable_result_is_still_returned(
380 provider: _FakeProvider, caplog: pytest.LogCaptureFixture
381) -> None:
382 """Test that callers still get a result when it cannot be copied."""
383 provider.gate.clear()
384 tasks = [asyncio.create_task(provider.fetch_uncopyable("a")) for _ in range(3)]
385 await _wait_for_flight(provider)
386 provider.gate.set()
387 results = await asyncio.gather(*tasks)
388 assert provider.calls == 1
389 assert all(isinstance(result, _Uncopyable) for result in results)
390 assert "Cannot copy the shared result" in caplog.text
391
392
393async def test_cancelled_fetch_cancels_every_caller(provider: _FakeProvider) -> None:
394 """Test that cancelling the shared fetch itself surfaces to all of its callers."""
395 provider.gate.clear()
396 tasks = [asyncio.create_task(provider.fetch("a")) for _ in range(3)]
397 await _wait_for_flight(provider)
398 provider.mass._tracked_tasks[f"cache_flight.{_PROVIDER}.fetch.a"].cancel()
399 results = await asyncio.gather(*tasks, return_exceptions=True)
400 assert all(isinstance(result, asyncio.CancelledError) for result in results)
401
402
403async def test_cancelled_caller_leaves_the_others_untouched(
404 cache_controller: CacheController, provider: _FakeProvider
405) -> None:
406 """Test that one caller giving up neither cancels the fetch nor the other callers."""
407 provider.result = "value"
408 provider.gate.clear()
409 tasks = [asyncio.create_task(provider.fetch("a")) for _ in range(3)]
410 await _wait_for_flight(provider)
411 tasks[0].cancel()
412 provider.gate.set()
413 assert await asyncio.gather(*tasks[1:]) == ["value", "value"]
414 assert tasks[0].cancelled()
415 assert provider.calls == 1
416 await _wait_for_stored(cache_controller, "fetch.a")
417
418
419async def test_sole_cancelled_caller_still_completes_the_fetch(
420 cache_controller: CacheController, provider: _FakeProvider
421) -> None:
422 """Test that a fetch runs to completion and stores after its only caller gave up."""
423 provider.result = "value"
424 provider.gate.clear()
425 task = asyncio.create_task(provider.fetch("a"))
426 await _wait_for_flight(provider)
427 task.cancel()
428 provider.gate.set()
429 with pytest.raises(asyncio.CancelledError):
430 await task
431 await _wait_for_stored(cache_controller, "fetch.a")
432 assert provider.calls == 1
433
434
435async def test_failing_fetch_logs_no_task_warning(
436 provider: _FakeProvider, caplog: pytest.LogCaptureFixture
437) -> None:
438 """Test that a raising fetch stays quiet, even once its only caller has gone."""
439 provider.error = MediaNotFoundError("not found")
440 provider.gate.clear()
441 task = asyncio.create_task(provider.fetch("a"))
442 await _wait_for_flight(provider)
443 task.cancel()
444 caplog.clear()
445 provider.gate.set()
446 with pytest.raises(asyncio.CancelledError):
447 await task
448 await asyncio.sleep(0.05)
449 assert [
450 record
451 for record in caplog.records
452 if record.levelno >= logging.WARNING and "Exception in task" in record.getMessage()
453 ] == []
454
455
456async def test_exception_is_shared_and_not_cached(
457 cache_controller: CacheController, provider: _FakeProvider
458) -> None:
459 """Test that every caller gets the raised error and the next call fetches again."""
460 provider.error = MediaNotFoundError("not found")
461 provider.gate.clear()
462 with patch.object(cache_controller, "set", AsyncMock()) as store:
463 tasks = [asyncio.create_task(provider.fetch("a")) for _ in range(3)]
464 await _wait_for_flight(provider)
465 provider.gate.set()
466 results = await asyncio.gather(*tasks, return_exceptions=True)
467 assert provider.calls == 1
468 assert all(result is provider.error for result in results)
469 assert store.await_count == 0
470 provider.error = None
471 provider.result = "value"
472 assert await provider.fetch("a") == "value"
473 assert provider.calls == 2
474
475
476async def test_cache_none_false_shares_one_attempt(
477 cache_controller: CacheController, provider: _FakeProvider
478) -> None:
479 """Test that concurrent callers share one attempt and one None, then retry after."""
480 provider.result = None
481 provider.gate.clear()
482 with patch.object(cache_controller, "set", AsyncMock()) as store:
483 tasks = [asyncio.create_task(provider.fetch_no_none("a")) for _ in range(3)]
484 await _wait_for_flight(provider)
485 provider.gate.set()
486 assert await asyncio.gather(*tasks) == [None, None, None]
487 assert provider.calls == 1
488 assert store.await_count == 0
489 assert await provider.fetch_no_none("a") is None
490 assert provider.calls == 2
491
492
493async def test_bypass_cache_does_not_join_a_fetch(provider: _FakeProvider) -> None:
494 """Test that a BYPASS_CACHE caller fetches on its own instead of joining."""
495
496 async def _bypassing() -> str | None:
497 token = BYPASS_CACHE.set(True)
498 try:
499 return await provider.fetch("a")
500 finally:
501 BYPASS_CACHE.reset(token)
502
503 provider.result = "value"
504 provider.gate.clear()
505 tasks = [asyncio.create_task(provider.fetch("a")), asyncio.create_task(_bypassing())]
506 await _wait_for_flight(provider)
507 provider.gate.set()
508 assert await asyncio.gather(*tasks) == ["value", "value"]
509 assert provider.calls == 2
510
511
512async def test_swr_refreshes_once_for_concurrent_callers(
513 cache_controller: CacheController, provider: _FakeProvider
514) -> None:
515 """Test that concurrent callers on a stale entry trigger one background refresh."""
516 await cache_controller.set(
517 "fetch_swr.a", "stale", provider=_PROVIDER, expiration=-1, allow_expired_cache=True
518 )
519 provider.result = "fresh"
520 provider.gate.clear()
521 tasks = [asyncio.create_task(provider.fetch_swr("a")) for _ in range(3)]
522 assert await asyncio.gather(*tasks) == ["stale", "stale", "stale"]
523 await _wait_for_flight(provider)
524 provider.gate.set()
525
526 async def _refreshed() -> bool:
527 return bool(await cache_controller.get("fetch_swr.a", provider=_PROVIDER) == "fresh")
528
529 await _wait_for(_refreshed)
530 assert provider.calls == 1
531
532
533async def test_completed_fetch_is_not_reused(
534 cache_controller: CacheController, provider: _FakeProvider
535) -> None:
536 """Test that a caller is not handed a fetch that already finished."""
537
538 async def _noop() -> None:
539 return None
540
541 # a finished flight left under the key must be replaced, not awaited for its outcome
542 finished = asyncio.create_task(_noop())
543 await finished
544 cache_controller.mass._tracked_tasks[f"cache_flight.{_PROVIDER}.fetch.a"] = finished
545 provider.result = "value"
546 assert await provider.fetch("a") == "value"
547 assert provider.calls == 1
548
549
550# --- get_with_freshness ---
551
552
553async def test_get_with_freshness(cache_controller: CacheController) -> None:
554 """Test that get_with_freshness reports the freshness and presence of entries."""
555 await cache_controller.set("fresh", "data", provider=_PROVIDER, expiration=3600)
556 assert await cache_controller.get_with_freshness("fresh", provider=_PROVIDER) == (
557 "data",
558 True,
559 True,
560 )
561 await cache_controller.set("expired", "old", provider=_PROVIDER, expiration=-1)
562 # an expired entry is reported as not found unless include_expired is set
563 assert await cache_controller.get_with_freshness("expired", provider=_PROVIDER) == (
564 None,
565 False,
566 False,
567 )
568 assert await cache_controller.get_with_freshness(
569 "expired", provider=_PROVIDER, include_expired=True
570 ) == (
571 "old",
572 False,
573 True,
574 )
575 data, is_fresh, found = await cache_controller.get_with_freshness("missing", provider=_PROVIDER)
576 assert found is False
577 assert is_fresh is False
578 assert data is None
579