music-assistant-server

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