/
/
/
1"""Tests for the source-image cache in the images helper."""
2
3from __future__ import annotations
4
5import asyncio
6import logging
7import os
8import subprocess
9import time
10from base64 import b64encode
11from io import BytesIO
12from pathlib import Path
13from typing import TYPE_CHECKING, Any
14from unittest.mock import AsyncMock, MagicMock
15
16import pytest
17from aiohttp import ClientSession, web
18from aiohttp.client_exceptions import ClientError
19from aiohttp.test_utils import TestServer
20from music_assistant_models.enums import ImageType, ProviderIconVariant
21from music_assistant_models.errors import MediaNotFoundError
22from music_assistant_models.media_items import MediaItemImage
23from PIL import Image
24
25from music_assistant.helpers import images
26from music_assistant.helpers.images import (
27 _SOURCE_CACHE_TTL,
28 create_thumb_hash,
29 detect_provider_icons,
30 get_image_data,
31 get_image_thumb,
32 get_image_thumb_path,
33 invalidate_cached_image,
34 load_provider_icon,
35)
36from music_assistant.models.metadata_provider import MetadataProvider
37from music_assistant.models.music_provider import MusicProvider
38from music_assistant.models.player_provider import PlayerProvider
39from tests.common import collect_loop_errors
40
41if TYPE_CHECKING:
42 from collections.abc import Iterator
43
44 from music_assistant.mass import MusicAssistant
45
46
47@pytest.fixture(autouse=True)
48def _reset_image_caches() -> Iterator[None]:
49 """Isolate the module-level image caches between tests."""
50 images._thumb_memory_cache.clear()
51 images._source_memory_cache.clear()
52 images._failed_sources.clear()
53 yield
54 images._thumb_memory_cache.clear()
55 images._source_memory_cache.clear()
56 images._failed_sources.clear()
57
58
59@pytest.fixture
60def fetch_calls(monkeypatch: pytest.MonkeyPatch) -> list[tuple[str, str]]:
61 """Spy on origin fetches; returns the list of (provider, path) fetch calls."""
62 calls: list[tuple[str, str]] = []
63 real_fetch = images._fetch_source_image
64
65 async def counting_fetch(
66 mass: MusicAssistant, path_or_url: str, provider: str, depth: int
67 ) -> tuple[bytes, bool]:
68 calls.append((provider, path_or_url))
69 return await real_fetch(mass, path_or_url, provider, depth)
70
71 monkeypatch.setattr(images, "_fetch_source_image", counting_fetch)
72 return calls
73
74
75def _make_png_bytes(color: tuple[int, int, int] = (200, 30, 30), size: int = 400) -> bytes:
76 """Create raw PNG bytes of a solid-color square."""
77 img = Image.new("RGB", (size, size), color)
78 buf = BytesIO()
79 img.save(buf, "PNG")
80 return buf.getvalue()
81
82
83def _make_png_file(tmp_path: Path, name: str = "art.png") -> str:
84 """Create a PNG file on disk and return its absolute path."""
85 filepath = tmp_path / name
86 filepath.write_bytes(_make_png_bytes())
87 return str(filepath)
88
89
90async def test_multiple_thumb_sizes_fetch_source_once(
91 mass_minimal: MusicAssistant, tmp_path: Path, fetch_calls: list[tuple[str, str]]
92) -> None:
93 """Generating several thumb variants of one image must fetch the source once."""
94 image_path = _make_png_file(tmp_path)
95 thumb_80 = await get_image_thumb(mass_minimal, image_path, 80, "builtin")
96 thumb_256 = await get_image_thumb(mass_minimal, image_path, 256, "builtin")
97 jpeg_flat = await get_image_thumb(
98 mass_minimal, image_path, 256, "builtin", image_format="JPEG", flatten_transparency=True
99 )
100 assert thumb_80
101 assert thumb_256
102 assert jpeg_flat
103 assert fetch_calls == [("builtin", image_path)]
104
105
106async def test_thumb_path_reuses_existing_cache_file(
107 mass_minimal: MusicAssistant, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
108) -> None:
109 """A repeated path lookup returns the same existing file without rewriting it."""
110 image_path = _make_png_file(tmp_path)
111 first_path = await get_image_thumb_path(mass_minimal, image_path, 256, "builtin")
112
113 async def unexpected_write(*_args: object, **_kwargs: object) -> None:
114 raise AssertionError("existing thumbnail was rewritten")
115
116 monkeypatch.setattr(images, "_write_thumb_to_disk", unexpected_write)
117 second_path = await get_image_thumb_path(mass_minimal, image_path, 256, "builtin")
118
119 assert second_path == first_path
120 assert Path(second_path).is_file()
121
122
123async def test_thumb_path_restores_missing_disk_file_from_memory(
124 mass_minimal: MusicAssistant, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
125) -> None:
126 """A memory hit recreates its missing disk file without regenerating the thumbnail."""
127 image_path = _make_png_file(tmp_path)
128 thumb_data = await get_image_thumb(mass_minimal, image_path, 256, "builtin")
129 thumb_hash = create_thumb_hash("builtin", image_path)
130 cache_filename = images._thumb_cache_filename(thumb_hash, 256, "PNG")
131 cache_path = Path(mass_minimal.cache_path, "thumbnails", cache_filename)
132 cache_path.unlink()
133
134 async def unexpected_generate(*_args: object, **_kwargs: object) -> bytes:
135 raise AssertionError("memory-cached thumbnail was regenerated")
136
137 monkeypatch.setattr(images, "_generate_and_cache_thumb", unexpected_generate)
138 restored_path = await get_image_thumb_path(mass_minimal, image_path, 256, "builtin")
139
140 assert restored_path == str(cache_path.resolve())
141 assert cache_path.read_bytes() == thumb_data
142
143
144async def test_thumb_path_surfaces_cache_write_error(
145 mass_minimal: MusicAssistant, tmp_path: Path
146) -> None:
147 """A path request raises when the thumbnail cache cannot be persisted."""
148 invalid_cache_path = tmp_path / "not-a-directory"
149 invalid_cache_path.write_text("file")
150 mass_minimal.cache_path = str(invalid_cache_path)
151 image_data = b64encode(_make_png_bytes()).decode()
152
153 with pytest.raises(NotADirectoryError, match="not-a-directory"):
154 await get_image_thumb_path(
155 mass_minimal,
156 f"data:image/png;base64,{image_data}",
157 256,
158 "builtin",
159 )
160
161
162async def test_concurrent_requests_share_one_fetch(
163 mass_minimal: MusicAssistant, monkeypatch: pytest.MonkeyPatch
164) -> None:
165 """Concurrent get_image_data calls for the same source coalesce into one fetch."""
166 gate = asyncio.Event()
167 calls: list[str] = []
168
169 async def gated_fetch(
170 _mass: MusicAssistant, path_or_url: str, _provider: str, _depth: int
171 ) -> tuple[bytes, bool]:
172 calls.append(path_or_url)
173 await gate.wait()
174 return b"image-bytes", False
175
176 monkeypatch.setattr(images, "_fetch_source_image", gated_fetch)
177 tasks = [
178 asyncio.create_task(get_image_data(mass_minimal, "/some/image.png", "builtin"))
179 for _ in range(3)
180 ]
181 await asyncio.sleep(0)
182 gate.set()
183 results = await asyncio.gather(*tasks)
184 assert results == [b"image-bytes"] * 3
185 assert calls == ["/some/image.png"]
186
187
188async def test_cancelled_caller_of_a_failing_fetch_logs_no_loop_error(
189 mass_minimal: MusicAssistant, monkeypatch: pytest.MonkeyPatch
190) -> None:
191 """A source fetch failing after a caller gave up is not reported to the loop handler."""
192 release = asyncio.Event()
193 calls: list[str] = []
194
195 async def failing_fetch(
196 _mass: MusicAssistant, path_or_url: str, _provider: str, _depth: int
197 ) -> tuple[bytes, bool]:
198 calls.append(path_or_url)
199 await release.wait()
200 raise FileNotFoundError(f"Image not found: {path_or_url}")
201
202 monkeypatch.setattr(images, "_fetch_source_image", failing_fetch)
203 with collect_loop_errors() as reported:
204 task_a = asyncio.create_task(get_image_data(mass_minimal, "/some/image.png", "builtin"))
205 task_b = asyncio.create_task(get_image_data(mass_minimal, "/some/image.png", "builtin"))
206 # let both callers await the (same) in-flight fetch, then cancel one
207 await asyncio.sleep(0)
208 task_a.cancel()
209 with pytest.raises(asyncio.CancelledError):
210 await task_a
211 # release the fetch only once the cancellation is fully processed, so the
212 # failure reliably lands after the giving-up caller is gone
213 release.set()
214 with pytest.raises(FileNotFoundError):
215 await task_b
216
217 assert calls == ["/some/image.png"]
218 assert reported == []
219
220
221async def test_cancelled_caller_of_a_failing_thumb_logs_no_loop_error(
222 mass_minimal: MusicAssistant, monkeypatch: pytest.MonkeyPatch
223) -> None:
224 """Thumbnail generation failing after its caller gave up is not reported either."""
225 entered = asyncio.Event()
226 release = asyncio.Event()
227 generation: list[asyncio.Task[Any]] = []
228
229 async def failing_source(_mass: MusicAssistant, path_or_url: str, _provider: str) -> bytes:
230 current = asyncio.current_task()
231 assert current is not None
232 generation.append(current)
233 entered.set()
234 await release.wait()
235 raise FileNotFoundError(f"Image not found: {path_or_url}")
236
237 monkeypatch.setattr(images, "get_image_data", failing_source)
238 with collect_loop_errors() as reported:
239 caller = asyncio.create_task(
240 get_image_thumb(mass_minimal, "/some/image.png", 256, "builtin")
241 )
242 await entered.wait()
243 caller.cancel()
244 with pytest.raises(asyncio.CancelledError):
245 await caller
246 # only fail the generation once the cancellation is fully processed
247 release.set()
248 await asyncio.wait(generation)
249
250 assert isinstance(generation[0].exception(), FileNotFoundError)
251 assert reported == []
252
253
254async def test_data_uri_is_decoded_without_caching(
255 mass_minimal: MusicAssistant, fetch_calls: list[tuple[str, str]]
256) -> None:
257 """Inline base64 data URIs are decoded directly and never enter the cache."""
258 payload = b"\x89PNG\r\n\x1a\nfakepngdata"
259 data_uri = f"data:image/png;base64,{b64encode(payload).decode()}"
260 result = await get_image_data(mass_minimal, data_uri, "builtin")
261 assert result == payload
262 assert fetch_calls == []
263 assert not images._source_memory_cache.entries
264
265
266async def test_provider_bytes_use_disk_cache_across_restart(
267 mass_minimal: MusicAssistant,
268 monkeypatch: pytest.MonkeyPatch,
269 fetch_calls: list[tuple[str, str]],
270) -> None:
271 """An expensive fetch is persisted on disk and reused after a (simulated) restart."""
272 fake_provider = MagicMock(spec=MetadataProvider)
273 fake_provider.resolve_image = AsyncMock(return_value=b"expensive-image-bytes")
274 monkeypatch.setattr(mass_minimal, "get_provider", lambda _prov: fake_provider)
275
276 data = await get_image_data(mass_minimal, "some/prov/path.jpg", "fake--1")
277 assert data == b"expensive-image-bytes"
278 assert len(fetch_calls) == 1
279 cache_key = create_thumb_hash("fake--1", "some/prov/path.jpg")
280 src_file = os.path.join(mass_minimal.cache_path, "thumbnails", f"{cache_key}_src")
281 assert Path(src_file).is_file()
282
283 # simulate a restart: memory tier gone, disk entry remains
284 images._source_memory_cache.clear()
285 data = await get_image_data(mass_minimal, "some/prov/path.jpg", "fake--1")
286 assert data == b"expensive-image-bytes"
287 assert len(fetch_calls) == 1
288
289
290async def test_player_provider_can_resolve_image_bytes(
291 mass_minimal: MusicAssistant,
292 monkeypatch: pytest.MonkeyPatch,
293) -> None:
294 """Player-provider images use the shared image retrieval pipeline."""
295 fake_provider = MagicMock(spec=PlayerProvider)
296 fake_provider.resolve_image = AsyncMock(return_value=b"player-image-bytes")
297 monkeypatch.setattr(mass_minimal, "get_provider", lambda _prov: fake_provider)
298
299 data = await get_image_data(mass_minimal, "player/artwork", "player--1")
300
301 assert data == b"player-image-bytes"
302 fake_provider.resolve_image.assert_awaited_once_with("player/artwork")
303
304
305async def test_local_file_read_cached_on_disk(
306 mass_minimal: MusicAssistant, tmp_path: Path, fetch_calls: list[tuple[str, str]]
307) -> None:
308 """A local file read lands in both cache tiers and is served from disk after restart."""
309 image_path = _make_png_file(tmp_path)
310 data = await get_image_data(mass_minimal, image_path, "builtin")
311 assert len(fetch_calls) == 1
312 cache_key = create_thumb_hash("builtin", image_path)
313 assert cache_key in images._source_memory_cache.entries
314 src_file = os.path.join(mass_minimal.cache_path, "thumbnails", f"{cache_key}_src")
315 assert Path(src_file).is_file()
316
317 # after a restart the disk entry serves the bytes without touching the origin
318 # (which may live on a network mount) - local entries have no TTL
319 images._source_memory_cache.clear()
320 assert await get_image_data(mass_minimal, image_path, "builtin") == data
321 assert len(fetch_calls) == 1
322
323
324async def test_remote_disk_entry_expires_after_ttl(
325 mass_minimal: MusicAssistant,
326 monkeypatch: pytest.MonkeyPatch,
327 fetch_calls: list[tuple[str, str]],
328) -> None:
329 """A stale on-disk entry for a remote url is refetched after the TTL."""
330 mass_minimal.webserver = MagicMock(base_url="http://127.0.0.1:8095")
331 mass_minimal.streams = MagicMock(base_url="http://127.0.0.1:8097")
332 remote_url = "http://cdn.example.com/artwork.jpg"
333
334 async def fake_remote_fetch(_mass: MusicAssistant, _url: str) -> bytes:
335 return b"remote-image-bytes"
336
337 monkeypatch.setattr(images, "_fetch_remote_image", fake_remote_fetch)
338 await get_image_data(mass_minimal, remote_url, "builtin")
339 assert len(fetch_calls) == 1
340 cache_key = create_thumb_hash("builtin", remote_url)
341 src_file = os.path.join(mass_minimal.cache_path, "thumbnails", f"{cache_key}_src")
342 assert Path(src_file).is_file()
343
344 # a fresh disk entry is used after a restart...
345 images._source_memory_cache.clear()
346 await get_image_data(mass_minimal, remote_url, "builtin")
347 assert len(fetch_calls) == 1
348 # ...but an expired one is not
349 images._source_memory_cache.clear()
350 expired = time.time() - _SOURCE_CACHE_TTL - 10
351 os.utime(src_file, (expired, expired))
352 await get_image_data(mass_minimal, remote_url, "builtin")
353 assert len(fetch_calls) == 2
354
355
356async def test_failing_source_fails_fast_with_single_warning(
357 mass_minimal: MusicAssistant,
358 monkeypatch: pytest.MonkeyPatch,
359 caplog: pytest.LogCaptureFixture,
360 fetch_calls: list[tuple[str, str]],
361) -> None:
362 """A persistently failing source is fetched once, then fails fast without new logs."""
363 mass_minimal.webserver = MagicMock(base_url="http://127.0.0.1:8095")
364 mass_minimal.streams = MagicMock(base_url="http://127.0.0.1:8097")
365 remote_url = "http://sonos.example.com:1400/getaa?u=missing.flac"
366
367 async def failing_remote_fetch(_mass: MusicAssistant, url: str) -> bytes:
368 raise ClientError(f"404, message='Not Found', url='{url}'")
369
370 monkeypatch.setattr(images, "_fetch_remote_image", failing_remote_fetch)
371 caplog.set_level(logging.WARNING, logger="music_assistant.helpers.images")
372
373 with pytest.raises(FileNotFoundError, match="404"):
374 await get_image_data(mass_minimal, remote_url, "builtin")
375 # follow-up requests (next metadata push, a thumbnail, a palette) fail
376 # fast without a new origin fetch and without logging again
377 with pytest.raises(FileNotFoundError, match="404"):
378 await get_image_data(mass_minimal, remote_url, "builtin")
379 with pytest.raises(FileNotFoundError, match="404"):
380 await get_image_thumb(mass_minimal, remote_url, 256, "builtin")
381
382 assert len(fetch_calls) == 1
383 warnings = [rec for rec in caplog.records if rec.name == "music_assistant.helpers.images"]
384 assert len(warnings) == 1
385 assert "not retrying" in warnings[0].getMessage()
386
387
388async def test_provider_reported_missing_image_fails_fast_with_single_warning(
389 mass_minimal: MusicAssistant,
390 monkeypatch: pytest.MonkeyPatch,
391 caplog: pytest.LogCaptureFixture,
392 fetch_calls: list[tuple[str, str]],
393) -> None:
394 """A provider reporting a missing image is asked once, then fails fast without new logs."""
395 fake_provider = MagicMock(spec=MusicProvider)
396 fake_provider.resolve_image = AsyncMock(
397 side_effect=MediaNotFoundError("Image path is a directory: Some Artist")
398 )
399 monkeypatch.setattr(mass_minimal, "get_provider", lambda _prov: fake_provider)
400 caplog.set_level(logging.WARNING, logger="music_assistant.helpers.images")
401
402 with pytest.raises(MediaNotFoundError, match="Some Artist"):
403 await get_image_data(mass_minimal, "Some Artist", "filesystem_local--1")
404 # follow-up requests (a thumbnail, a palette) fail fast from the negative cache,
405 # without asking the provider again and without logging again
406 with pytest.raises(FileNotFoundError, match="Some Artist"):
407 await get_image_data(mass_minimal, "Some Artist", "filesystem_local--1")
408
409 assert len(fetch_calls) == 1
410 assert fake_provider.resolve_image.await_count == 1
411 warnings = [rec for rec in caplog.records if rec.name == "music_assistant.helpers.images"]
412 assert len(warnings) == 1
413 assert "not retrying" in warnings[0].getMessage()
414
415
416async def test_failed_source_retried_after_ttl_or_invalidation(
417 mass_minimal: MusicAssistant,
418 monkeypatch: pytest.MonkeyPatch,
419 fetch_calls: list[tuple[str, str]],
420) -> None:
421 """A failed source is retried after the negative-cache TTL or invalidation."""
422 mass_minimal.webserver = MagicMock(base_url="http://127.0.0.1:8095")
423 mass_minimal.streams = MagicMock(base_url="http://127.0.0.1:8097")
424 remote_url = "http://cdn.example.com/broken.jpg"
425 cache_key = create_thumb_hash("builtin", remote_url)
426
427 async def failing_remote_fetch(_mass: MusicAssistant, _url: str) -> bytes:
428 raise ClientError("503, message='Service Unavailable'")
429
430 monkeypatch.setattr(images, "_fetch_remote_image", failing_remote_fetch)
431 with pytest.raises(FileNotFoundError):
432 await get_image_data(mass_minimal, remote_url, "builtin")
433 assert len(fetch_calls) == 1
434
435 # once the TTL has passed, the origin is tried again
436 _expires_at, message = images._failed_sources[cache_key]
437 images._failed_sources[cache_key] = (time.monotonic() - 1, message)
438 with pytest.raises(FileNotFoundError):
439 await get_image_data(mass_minimal, remote_url, "builtin")
440 assert len(fetch_calls) == 2
441
442 # invalidation drops the entry immediately; a recovered origin then serves
443 await invalidate_cached_image(mass_minimal, "builtin", remote_url)
444 assert cache_key not in images._failed_sources
445
446 async def ok_remote_fetch(_mass: MusicAssistant, _url: str) -> bytes:
447 return b"remote-image-bytes"
448
449 monkeypatch.setattr(images, "_fetch_remote_image", ok_remote_fetch)
450 assert await get_image_data(mass_minimal, remote_url, "builtin") == b"remote-image-bytes"
451 assert len(fetch_calls) == 3
452
453
454async def test_remote_http_404_yields_file_not_found(mass_minimal: MusicAssistant) -> None:
455 """A real HTTP 404 response converts into FileNotFoundError with one origin hit."""
456 hits = 0
457
458 async def handler(_request: web.Request) -> web.Response:
459 nonlocal hits
460 hits += 1
461 return web.Response(status=404)
462
463 app = web.Application()
464 app.router.add_get("/getaa", handler)
465 server = TestServer(app)
466 await server.start_server()
467 session = ClientSession()
468 try:
469 mass_minimal.webserver = MagicMock(base_url="http://127.0.0.1:8095")
470 mass_minimal.streams = MagicMock(base_url="http://127.0.0.1:8097")
471 mass_minimal._http_session_no_ssl = session
472 url = str(server.make_url("/getaa")) + "?u=track.flac"
473 with pytest.raises(FileNotFoundError, match="404"):
474 await get_image_data(mass_minimal, url, "builtin")
475 # the negative cache prevents a second hit on the origin
476 with pytest.raises(FileNotFoundError, match="404"):
477 await get_image_data(mass_minimal, url, "builtin")
478 assert hits == 1
479 finally:
480 await session.close()
481 await server.close()
482
483
484async def test_own_imageproxy_url_cached_under_resolved_key_only(
485 mass_minimal: MusicAssistant, tmp_path: Path, fetch_calls: list[tuple[str, str]]
486) -> None:
487 """Imageproxy URLs to our own server never create an alias cache entry."""
488 image_path = _make_png_file(tmp_path)
489 mass_minimal.webserver = MagicMock(base_url="http://127.0.0.1:8095")
490 mass_minimal.streams = MagicMock(base_url="http://127.0.0.1:8097")
491 image_id = create_thumb_hash("builtin", image_path)
492 mass_minimal.metadata = MagicMock(
493 resolve_image_id=AsyncMock(return_value=("builtin", image_path))
494 )
495 proxy_url = f"http://127.0.0.1:8095/imageproxy/{image_id}?size=256"
496
497 await get_image_data(mass_minimal, proxy_url, "builtin")
498 assert fetch_calls == [("builtin", image_path)]
499 assert create_thumb_hash("builtin", image_path) in images._source_memory_cache.entries
500 assert create_thumb_hash("builtin", proxy_url) not in images._source_memory_cache.entries
501 # a repeated request through the proxy url form hits the resolved cache entry
502 await get_image_data(mass_minimal, proxy_url, "builtin")
503 assert len(fetch_calls) == 1
504
505
506async def test_invalidate_cached_image_clears_all_tiers(
507 mass_minimal: MusicAssistant, tmp_path: Path, fetch_calls: list[tuple[str, str]]
508) -> None:
509 """Invalidation removes every thumb variant plus source bytes, then refetches."""
510 image_path = _make_png_file(tmp_path, "target.png")
511 other_path = _make_png_file(tmp_path, "other.png")
512 for path in (image_path, other_path):
513 await get_image_thumb(mass_minimal, path, 80, "builtin")
514 await get_image_thumb(mass_minimal, path, 256, "builtin")
515 assert len(fetch_calls) == 2
516
517 thumb_dir = Path(mass_minimal.cache_path, "thumbnails")
518 target_hash = create_thumb_hash("builtin", image_path)
519 other_hash = create_thumb_hash("builtin", other_path)
520 # two thumb variants plus the source entry per image
521 assert len([f for f in thumb_dir.iterdir() if f.name.startswith(target_hash)]) == 3
522
523 await invalidate_cached_image(mass_minimal, "builtin", image_path)
524
525 # all artifacts of the target image are gone, the other image is untouched
526 assert not [f for f in thumb_dir.iterdir() if f.name.startswith(target_hash)]
527 assert len([f for f in thumb_dir.iterdir() if f.name.startswith(other_hash)]) == 3
528 assert target_hash not in images._source_memory_cache.entries
529 assert not any(key.startswith(f"{target_hash}_") for key in images._thumb_memory_cache)
530 assert any(key.startswith(f"{other_hash}_") for key in images._thumb_memory_cache)
531
532 # the next request must hit the origin again
533 await get_image_thumb(mass_minimal, image_path, 80, "builtin")
534 assert len(fetch_calls) == 3
535
536
537async def test_source_memory_cache_respects_byte_budget(
538 monkeypatch: pytest.MonkeyPatch,
539) -> None:
540 """The memory tier evicts oldest entries once the byte budget is exceeded."""
541 monkeypatch.setattr(images, "_SOURCE_MEMORY_MAX_BYTES", 1000)
542 monkeypatch.setattr(images, "_SOURCE_MEMORY_ENTRY_MAX_BYTES", 600)
543 cache = images._source_memory_cache
544 cache.put("aa", b"x" * 400)
545 cache.put("bb", b"y" * 400)
546 assert cache.get("aa") is not None
547 # third entry pushes total over budget: oldest entry is evicted
548 cache.put("cc", b"z" * 400)
549 assert cache.get("bb") is None
550 assert cache.get("aa") is not None # refreshed by the get above
551 assert cache.get("cc") is not None
552 # an entry larger than the per-entry cap is not stored at all
553 cache.put("dd", b"w" * 601)
554 assert cache.get("dd") is None
555 assert cache.total_bytes == 800
556
557
558async def test_source_memory_cache_entries_expire(monkeypatch: pytest.MonkeyPatch) -> None:
559 """Memory entries older than the TTL are treated as a miss."""
560 cache = images._source_memory_cache
561 cache.put("aa", b"payload")
562 assert cache.get("aa") == b"payload"
563 # shrink the TTL so the stored entry is immediately considered expired
564 monkeypatch.setattr(images, "_SOURCE_CACHE_TTL", -1)
565 assert cache.get("aa") is None
566 assert cache.total_bytes == 0
567
568
569def _create_mp3_with_cover(track_path: str, cover_path: str) -> None:
570 """Create a short mp3 file with the given image embedded as cover art."""
571 subprocess.run( # noqa: S603
572 [ # noqa: S607
573 "ffmpeg",
574 "-y",
575 "-hide_banner",
576 "-loglevel",
577 "error",
578 "-f",
579 "lavfi",
580 "-t",
581 "0.1",
582 "-i",
583 "anullsrc=r=44100:cl=mono",
584 "-i",
585 cover_path,
586 "-map",
587 "0:a",
588 "-map",
589 "1:v",
590 "-c:a",
591 "libmp3lame",
592 "-c:v",
593 "mjpeg",
594 "-id3v2_version",
595 "3",
596 track_path,
597 ],
598 check=True,
599 capture_output=True,
600 )
601
602
603async def test_embedded_art_retag_flow(
604 mass_minimal: MusicAssistant, tmp_path: Path, fetch_calls: list[tuple[str, str]]
605) -> None:
606 """
607 The full retag scenario against a real audio file.
608
609 Embedded art (an expensive ffmpeg extraction) is cached in memory and on
610 disk; replacing the artwork in the file goes unnoticed until the cache is
611 invalidated, after which the new artwork is extracted and served.
612 """
613 red_cover = str(tmp_path / "red.png")
614 blue_cover = str(tmp_path / "blue.png")
615 Image.new("RGB", (64, 64), (255, 0, 0)).save(red_cover, "PNG")
616 Image.new("RGB", (64, 64), (0, 0, 255)).save(blue_cover, "PNG")
617 track_path = str(tmp_path / "track.mp3")
618 _create_mp3_with_cover(track_path, red_cover)
619
620 original_art = await get_image_data(mass_minimal, track_path, "builtin")
621 assert original_art.startswith(b"\xff\xd8") # extracted as JPEG
622 assert len(fetch_calls) == 1
623 cache_key = create_thumb_hash("builtin", track_path)
624 src_file = os.path.join(mass_minimal.cache_path, "thumbnails", f"{cache_key}_src")
625 assert Path(src_file).is_file() # ffmpeg extraction is disk-cache worthy
626
627 # a restart later, the disk entry avoids re-running ffmpeg
628 images._source_memory_cache.clear()
629 assert await get_image_data(mass_minimal, track_path, "builtin") == original_art
630 assert len(fetch_calls) == 1
631
632 # "retag" the file with new artwork: the cache still serves the old art...
633 _create_mp3_with_cover(track_path, blue_cover)
634 assert await get_image_data(mass_minimal, track_path, "builtin") == original_art
635 assert len(fetch_calls) == 1
636 # ...until it is invalidated, after which the new art is extracted
637 await invalidate_cached_image(mass_minimal, "builtin", track_path)
638 assert not Path(src_file).exists()
639 new_art = await get_image_data(mass_minimal, track_path, "builtin")
640 assert len(fetch_calls) == 2
641 assert new_art != original_art
642
643
644async def test_create_collage_fetches_each_unique_image_once(
645 mass_minimal: MusicAssistant, tmp_path: Path, fetch_calls: list[tuple[str, str]]
646) -> None:
647 """A collage fetches each unique image once and skips unfetchable ones."""
648 paths = [
649 str((tmp_path / name).absolute())
650 for name in ("one.png", "two.png", "three.png", "missing.png")
651 ]
652 for path, color in zip(paths[:3], ((200, 30, 30), (30, 200, 30), (30, 30, 200)), strict=True):
653 Image.new("RGB", (300, 300), color).save(path, "PNG")
654 collage_images = [
655 MediaItemImage(
656 type=ImageType.THUMB, path=path, provider="builtin", remotely_accessible=False
657 )
658 for path in paths
659 ]
660 collage = await images.create_collage(mass_minimal, collage_images, dimensions=(500, 500))
661 assert collage.startswith(b"\xff\xd8") # JPEG magic
662 # each unique image was fetched exactly once (incl. the one failed attempt)
663 assert sorted(path for _prov, path in fetch_calls) == sorted(paths)
664
665
666# Provider icon helpers tests
667async def test_load_provider_icon_svg(tmp_path: Path) -> None:
668 """Test loading an SVG icon file returns minified UTF-8 bytes."""
669 icon = tmp_path / "icon.svg"
670 icon.write_text("<svg>\n <path/>\n</svg>\n")
671 mime, data = await load_provider_icon(str(icon))
672 assert mime == "image/svg+xml"
673 assert data == b"<svg> <path/></svg>"
674
675
676async def test_load_provider_icon_png(tmp_path: Path) -> None:
677 """Test loading a PNG icon file returns raw bytes."""
678 icon = tmp_path / "icon.png"
679 raw = b"\x89PNG\r\n\x1a\n\x00\x00"
680 icon.write_bytes(raw)
681 mime, data = await load_provider_icon(str(icon))
682 assert mime == "image/png"
683 assert data == raw
684
685
686async def test_load_provider_icon_bad_ext(tmp_path: Path) -> None:
687 """Test loading an unsupported file format raises ValueError."""
688 icon = tmp_path / "icon.gif"
689 icon.write_bytes(b"gif")
690 with pytest.raises(ValueError, match="Unsupported"):
691 await load_provider_icon(str(icon))
692
693
694async def test_detect_provider_icons_svg_preferred(tmp_path: Path) -> None:
695 """Test that SVG is preferred over PNG when both exist."""
696 # both svg and png present for default -> svg wins
697 (tmp_path / "icon.svg").write_text("<svg/>")
698 (tmp_path / "icon.png").write_bytes(b"PNGDATA")
699 icons = await detect_provider_icons(str(tmp_path))
700 assert icons[ProviderIconVariant.DEFAULT] == ("image/svg+xml", b"<svg/>")
701 assert set(icons) == {ProviderIconVariant.DEFAULT}
702
703
704async def test_detect_provider_icons_png_and_variants(tmp_path: Path) -> None:
705 """Test detecting multiple icon variants."""
706 (tmp_path / "icon.png").write_bytes(b"PNGDATA")
707 (tmp_path / "icon_dark.svg").write_text("<svg/>")
708 (tmp_path / "icon_monochrome.png").write_bytes(b"MONO")
709 icons = await detect_provider_icons(str(tmp_path))
710 assert icons[ProviderIconVariant.DEFAULT] == ("image/png", b"PNGDATA")
711 assert icons[ProviderIconVariant.DARK] == ("image/svg+xml", b"<svg/>")
712 assert icons[ProviderIconVariant.MONOCHROME] == ("image/png", b"MONO")
713
714
715async def test_detect_provider_icons_none(tmp_path: Path) -> None:
716 """Test detecting icons in an empty directory returns empty dict."""
717 assert await detect_provider_icons(str(tmp_path)) == {}
718