music-assistant-server

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