music-assistant-server

1"""Tests for the imageproxy id system on the MetaDataController."""
2
3import asyncio
4import hashlib
5from pathlib import Path
6from typing import Any
7from unittest.mock import MagicMock
8
9import pytest
10from aiohttp import web
11from music_assistant_models.enums import ImageType
12from music_assistant_models.media_items import (
13    MediaItemImage,
14    MediaItemMetadata,
15    ProviderMapping,
16    Track,
17)
18from music_assistant_models.media_items.metadata import IMAGE_PROXY_ID_RESOLVER
19from music_assistant_models.unique_list import UniqueList
20from PIL import Image
21
22from music_assistant.controllers.metadata import MetaDataController
23from music_assistant.controllers.metadata.constants import (
24    _IMAGE_ID_CACHE_TTL,
25    _IMAGEPROXY_CONTENT_TYPES,
26    CACHE_CATEGORY_IMAGE_IDS,
27)
28from music_assistant.controllers.metadata.helpers import (
29    _normalize_imageproxy_format,
30)
31from music_assistant.helpers import images as images_helper
32from music_assistant.helpers.colors import get_palette
33from music_assistant.helpers.images import (
34    _THUMB_CACHE_VERSION,
35    _THUMB_FILENAME_RE,
36    _extract_imageproxy_id,
37    _has_alpha,
38    _thumb_cache_filename,
39    create_thumb_hash,
40    detect_image_content_format,
41    get_image_thumb,
42    is_svg_data,
43    player_image_url,
44)
45
46
47@pytest.fixture
48async def metadata_controller(
49    cache_database: None,  # noqa: ARG001
50    metadata_controller: MetaDataController,
51) -> MetaDataController:
52    """
53    Construct a MetaDataController backed by the cache database.
54
55    The controller tests in this module all persist and resolve image ids, unlike the
56    pure helper tests here, which need neither a controller nor a cache database.
57    """
58    return metadata_controller
59
60
61async def _wait_for_persisted_image_id(
62    controller: MetaDataController, image_id: str, timeout: float = 2.0
63) -> dict[str, str]:
64    """
65    Wait until `_persist_image_id` has written its row, or fail the test.
66
67    `_persist_image_id` runs on the loop but does real work (json dump on a
68    thread + sqlite write), so a `sleep(0)` busy-loop can race on slow CI.
69    """
70    loop = asyncio.get_running_loop()
71    deadline = loop.time() + timeout
72    while loop.time() < deadline:
73        raw = await controller.cache.get(
74            key=image_id,
75            category=CACHE_CATEGORY_IMAGE_IDS,
76            provider=controller.domain,
77        )
78        if raw is not None:
79            assert isinstance(raw, dict)
80            return raw
81        await asyncio.sleep(0.01)
82    raise AssertionError(f"image_id {image_id!r} not persisted within {timeout}s")
83
84
85async def test_compute_image_id_is_deterministic(metadata_controller: MetaDataController) -> None:
86    """The same (provider, path) must always produce the same hex id."""
87    image_id_a = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
88    image_id_b = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
89    assert image_id_a == image_id_b
90    expected = hashlib.sha256(b"filesystem//local/cover.jpg").hexdigest()
91    assert image_id_a == expected
92
93
94async def test_image_id_matches_thumb_and_palette_cache_key(
95    metadata_controller: MetaDataController,
96) -> None:
97    """
98    image_id must equal create_thumb_hash(provider, path).
99
100    The thumbnail and color-palette caches both key off
101    `create_thumb_hash(provider, path)`. If `compute_image_id` ever diverged
102    from that, palette lookups for /imageproxy/<id> URLs would miss and the
103    on-disk thumbnail cache would split into two buckets.
104    """
105    image_id = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
106    assert image_id == create_thumb_hash("filesystem", "/local/cover.jpg")
107
108
109async def test_compute_image_id_differs_per_input(metadata_controller: MetaDataController) -> None:
110    """Different (provider, path) pairs must produce different ids."""
111    assert metadata_controller.compute_image_id("x", "/a") != metadata_controller.compute_image_id(
112        "x", "/b"
113    )
114    assert metadata_controller.compute_image_id("x", "/a") != metadata_controller.compute_image_id(
115        "y", "/a"
116    )
117
118
119async def test_resolve_image_id_via_in_memory_lru(
120    metadata_controller: MetaDataController,
121) -> None:
122    """A freshly computed id resolves from the in-memory LRU without hitting DB."""
123    image_id = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
124    resolved = await metadata_controller.resolve_image_id(image_id)
125    assert resolved == ("filesystem", "/local/cover.jpg")
126
127
128async def test_resolve_image_id_via_cache_db(metadata_controller: MetaDataController) -> None:
129    """After the async persist runs, the mapping resolves from cache even with empty LRU."""
130    image_id = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
131    raw = await _wait_for_persisted_image_id(metadata_controller, image_id)
132    assert raw == {"provider": "filesystem", "path": "/local/cover.jpg"}
133    # wipe the in-memory layer so we hit the cache db
134    metadata_controller._image_id_lru.clear()
135    resolved = await metadata_controller.resolve_image_id(image_id)
136    assert resolved == ("filesystem", "/local/cover.jpg")
137
138
139async def test_resolve_image_id_returns_none_for_unknown(
140    metadata_controller: MetaDataController,
141) -> None:
142    """An id that was never registered must resolve to None (→ 404)."""
143    resolved = await metadata_controller.resolve_image_id("0" * 64)
144    assert resolved is None
145
146
147async def test_image_id_persists_with_persistent_flag(
148    metadata_controller: MetaDataController,
149) -> None:
150    """Mappings must survive a `clear()` without include_persistent."""
151    image_id = metadata_controller.compute_image_id("filesystem", "/persistent.jpg")
152    await _wait_for_persisted_image_id(metadata_controller, image_id)
153    # simulate the user-facing "Reset cache" action: clear() without include_persistent
154    await metadata_controller.cache.clear()
155    metadata_controller._image_id_lru.clear()
156    resolved = await metadata_controller.resolve_image_id(image_id)
157    assert resolved == ("filesystem", "/persistent.jpg")
158
159
160def _track_with_image(item_id: str, image: MediaItemImage) -> Track:
161    """Build a minimal serializable Track carrying the given image."""
162    return Track(
163        item_id=item_id,
164        provider="library",
165        name=f"Track {item_id}",
166        provider_mappings={
167            ProviderMapping(
168                item_id=item_id,
169                provider_domain="filesystem",
170                provider_instance="filesystem--test",
171            )
172        },
173        metadata=MediaItemMetadata(images=UniqueList([image])),
174    )
175
176
177async def _wait_for_expiration_refresh(
178    controller: MetaDataController, image_id: str, old_expires: int, timeout: float = 2.0
179) -> int:
180    """Wait until the stored row's expiration moved past `old_expires`, or fail."""
181    loop = asyncio.get_running_loop()
182    deadline = loop.time() + timeout
183    while loop.time() < deadline:
184        expires = await controller.cache.get_expiration(
185            key=image_id,
186            category=CACHE_CATEGORY_IMAGE_IDS,
187            provider=controller.domain,
188        )
189        if expires is not None and expires > old_expires:
190            return expires
191        await asyncio.sleep(0.01)
192    raise AssertionError(f"expiration for {image_id!r} not refreshed within {timeout}s")
193
194
195async def test_serialize_twice_hashes_once_and_persists_once(
196    metadata_controller: MetaDataController, monkeypatch: pytest.MonkeyPatch
197) -> None:
198    """
199    Serializing the same item list twice hashes and persists each unique image only once.
200
201    The proxy_id injection runs per image occurrence per outbound message, so
202    repeat serializations must be served from the forward memo: no hashing at
203    all on the second pass and exactly one cache write per unique image.
204    """
205    hash_calls: list[tuple[str, str]] = []
206
207    def counting_thumb_hash(provider: str, path: str) -> str:
208        hash_calls.append((provider, path))
209        return create_thumb_hash(provider, path)
210
211    monkeypatch.setattr(
212        "music_assistant.controllers.metadata.images.create_thumb_hash", counting_thumb_hash
213    )
214    persist_calls: list[str] = []
215    real_cache_set = metadata_controller.cache.set
216
217    async def counting_cache_set(*args: Any, **kwargs: Any) -> None:
218        if kwargs.get("category") == CACHE_CATEGORY_IMAGE_IDS:
219            persist_calls.append(kwargs["key"])
220        await real_cache_set(*args, **kwargs)
221
222    monkeypatch.setattr(metadata_controller.cache, "set", counting_cache_set)
223
224    # three image occurrences per pass, but only two unique images
225    image_a = MediaItemImage(type=ImageType.THUMB, path="/album_a.jpg", provider="filesystem")
226    image_b = MediaItemImage(type=ImageType.THUMB, path="/album_b.jpg", provider="filesystem")
227    tracks = [
228        _track_with_image("1", image_a),
229        _track_with_image("2", image_a),
230        _track_with_image("3", image_b),
231    ]
232    token = IMAGE_PROXY_ID_RESOLVER.set(metadata_controller.compute_image_id)
233    try:
234        first_pass = [track.to_dict() for track in tracks]
235        assert len(hash_calls) == 2
236        second_pass = [track.to_dict() for track in tracks]
237    finally:
238        IMAGE_PROXY_ID_RESOLVER.reset(token)
239    # zero hashing on the second pass: everything came from the forward memo
240    assert len(hash_calls) == 2
241    assert first_pass == second_pass
242    expected_ids = sorted(
243        create_thumb_hash("filesystem", path) for path in ("/album_a.jpg", "/album_b.jpg")
244    )
245    for track_dict in first_pass:
246        assert track_dict["metadata"]["images"][0]["proxy_id"] in expected_ids
247    # exactly one persist per unique image across both passes
248    for image_id in expected_ids:
249        await _wait_for_persisted_image_id(metadata_controller, image_id)
250    assert sorted(persist_calls) == expected_ids
251
252
253async def test_persist_skipped_when_row_already_fresh(
254    metadata_controller: MetaDataController, monkeypatch: pytest.MonkeyPatch
255) -> None:
256    """A mapping already stored with plenty of TTL left must not be rewritten."""
257    provider, path = "filesystem", "/prestored.jpg"
258    image_id = create_thumb_hash(provider, path)
259    # simulate a previous process run that already stored the mapping
260    await metadata_controller.cache.set(
261        key=image_id,
262        data={"provider": provider, "path": path},
263        category=CACHE_CATEGORY_IMAGE_IDS,
264        provider=metadata_controller.domain,
265        expiration=_IMAGE_ID_CACHE_TTL,
266        persistent=True,
267    )
268    probes: list[str] = []
269    real_get_expiration = metadata_controller.cache.get_expiration
270
271    async def recording_get_expiration(*args: Any, **kwargs: Any) -> int | None:
272        result = await real_get_expiration(*args, **kwargs)
273        probes.append(kwargs.get("key") or args[0])
274        return result
275
276    writes: list[str] = []
277    real_cache_set = metadata_controller.cache.set
278
279    async def recording_cache_set(*args: Any, **kwargs: Any) -> None:
280        writes.append(kwargs.get("key") or args[0])
281        await real_cache_set(*args, **kwargs)
282
283    monkeypatch.setattr(metadata_controller.cache, "get_expiration", recording_get_expiration)
284    monkeypatch.setattr(metadata_controller.cache, "set", recording_cache_set)
285
286    assert metadata_controller.compute_image_id(provider, path) == image_id
287    # wait for the freshness probe, then give the persist task time to (not) write
288    loop = asyncio.get_running_loop()
289    deadline = loop.time() + 2.0
290    while not probes and loop.time() < deadline:
291        await asyncio.sleep(0.01)
292    assert probes == [image_id]
293    await asyncio.sleep(0.05)
294    assert writes == []
295    # the pre-stored row keeps the id resolvable, also without the in-memory layer
296    metadata_controller._image_id_lru.clear()
297    assert await metadata_controller.resolve_image_id(image_id) == (provider, path)
298
299
300async def test_persist_refreshes_stale_row(metadata_controller: MetaDataController) -> None:
301    """A stored row past half its TTL is rewritten so the id stays resolvable."""
302    provider, path = "filesystem", "/stale.jpg"
303    image_id = create_thumb_hash(provider, path)
304    # a row whose remaining TTL is way below half of _IMAGE_ID_CACHE_TTL
305    await metadata_controller.cache.set(
306        key=image_id,
307        data={"provider": provider, "path": path},
308        category=CACHE_CATEGORY_IMAGE_IDS,
309        provider=metadata_controller.domain,
310        expiration=1000,
311        persistent=True,
312    )
313    old_expires = await metadata_controller.cache.get_expiration(
314        key=image_id,
315        category=CACHE_CATEGORY_IMAGE_IDS,
316        provider=metadata_controller.domain,
317    )
318    assert old_expires is not None
319    assert metadata_controller.compute_image_id(provider, path) == image_id
320    await _wait_for_expiration_refresh(metadata_controller, image_id, old_expires)
321
322
323async def test_imageproxy_resolves_after_restart(
324    metadata_controller: MetaDataController, monkeypatch: pytest.MonkeyPatch
325) -> None:
326    """
327    An id persisted by one process run must resolve on the imageproxy of the next.
328
329    Simulates a restart with a second controller instance on the same cache
330    database: cold in-memory maps, so resolution can only come from the
331    persisted row.
332    """
333    provider, path = "filesystem", "/restart/cover.jpg"
334    image_id = metadata_controller.compute_image_id(provider, path)
335    await _wait_for_persisted_image_id(metadata_controller, image_id)
336
337    restarted = MetaDataController(metadata_controller.mass)
338    assert not restarted._image_id_forward
339    assert not restarted._image_id_lru
340    served: list[tuple[str, str]] = []
341
342    async def fake_serve_thumbnail(
343        path_arg: str, provider_arg: str, _size: int, _image_format: str
344    ) -> web.Response:
345        served.append((provider_arg, path_arg))
346        return web.Response(status=200)
347
348    monkeypatch.setattr(restarted, "_serve_thumbnail", fake_serve_thumbnail)
349    request = MagicMock()
350    request.path = f"/imageproxy/{image_id}"
351    request.query = {}
352    response = await restarted.handle_imageproxy(request)
353    assert response.status == 200
354    assert served == [(provider, path)]
355
356
357async def test_no_repersist_after_in_memory_eviction(
358    metadata_controller: MetaDataController, monkeypatch: pytest.MonkeyPatch
359) -> None:
360    """Re-encountering an image after memo/LRU eviction must not hit the cache db again."""
361    provider, path = "filesystem", "/evicted.jpg"
362    image_id = metadata_controller.compute_image_id(provider, path)
363    await _wait_for_persisted_image_id(metadata_controller, image_id)
364    # simulate LRU pressure evicting the in-memory maps (markers survive)
365    metadata_controller._image_id_forward.clear()
366    metadata_controller._image_id_lru.clear()
367
368    calls: list[str] = []
369
370    async def recording_get_expiration(*_args: object, **_kwargs: object) -> int | None:
371        calls.append("probe")
372        return None
373
374    async def recording_cache_set(*_args: object, **_kwargs: object) -> None:
375        calls.append("write")
376
377    monkeypatch.setattr(metadata_controller.cache, "get_expiration", recording_get_expiration)
378    monkeypatch.setattr(metadata_controller.cache, "set", recording_cache_set)
379
380    # re-encounter re-hashes (memo was evicted) but the persist marker is fresh,
381    # so no persist work is scheduled at all
382    assert metadata_controller.compute_image_id(provider, path) == image_id
383    await asyncio.sleep(0.05)
384    assert calls == []
385
386
387def test_normalize_imageproxy_format() -> None:
388    """Client `fmt` values are lowercased / trimmed and validated."""
389    assert _normalize_imageproxy_format("jpg") == "jpg"
390    assert _normalize_imageproxy_format("JPG") == "jpg"
391    assert _normalize_imageproxy_format("jpeg") == "jpeg"
392    assert _normalize_imageproxy_format("PNG") == "png"
393    assert _normalize_imageproxy_format("svg") == "svg"
394    assert _normalize_imageproxy_format("  png  ") == "png"
395    # invalid / empty input falls through so caller can detect from path
396    assert _normalize_imageproxy_format("") is None
397    assert _normalize_imageproxy_format(None) is None
398    assert _normalize_imageproxy_format("gif") is None
399    assert _normalize_imageproxy_format("bmp") is None
400
401
402def test_imageproxy_content_types_are_standards_compliant() -> None:
403    """Every accepted fmt must produce a standards-compliant MIME type."""
404    assert _IMAGEPROXY_CONTENT_TYPES["jpg"] == "image/jpeg"
405    assert _IMAGEPROXY_CONTENT_TYPES["jpeg"] == "image/jpeg"
406    assert _IMAGEPROXY_CONTENT_TYPES["png"] == "image/png"
407    assert _IMAGEPROXY_CONTENT_TYPES["svg"] == "image/svg+xml"
408
409
410def test_extract_imageproxy_id_matches_path_only() -> None:
411    """Only URLs whose path begins with /imageproxy/ should yield an id."""
412    valid = "http://mass.local/imageproxy/" + "a" * 64 + "?size=256"
413    assert _extract_imageproxy_id(valid) == "a" * 64
414    # an id-shaped substring inside a query string must not match
415    decoy = "http://other.example.com/foo?next=/imageproxy/" + "b" * 64
416    assert _extract_imageproxy_id(decoy) is None
417    # invalid id length / charset
418    assert _extract_imageproxy_id("http://mass.local/imageproxy/short") is None
419    assert _extract_imageproxy_id("http://mass.local/imageproxy/" + "g" * 64) is None
420    # extra path segments must not be accepted (matches handle_imageproxy)
421    assert _extract_imageproxy_id("http://mass.local/imageproxy/" + "a" * 64 + "/extra") is None
422    assert _extract_imageproxy_id("http://mass.local/imageproxy/extra/" + "a" * 64) is None
423    # canonical form with trailing slash is fine
424    assert _extract_imageproxy_id("http://mass.local/imageproxy/" + "a" * 64 + "/") == "a" * 64
425
426
427async def test_handle_imageproxy_rejects_extra_path_segments(
428    metadata_controller: MetaDataController,
429) -> None:
430    """`/imageproxy/<id>` is the only shape that should validate."""
431    image_id = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
432
433    def _fake_request(path: str) -> MagicMock:
434        request = MagicMock()
435        request.path = path
436        request.query = {}
437        return request
438
439    # canonical form passes the path check and reaches resolve (404 because
440    # the (provider, path) is not actually fetchable in this minimal fixture)
441    assert (
442        await metadata_controller.handle_imageproxy(_fake_request(f"/imageproxy/{image_id}"))
443    ).status in (200, 404)
444    # extra leading segment must not be accepted
445    bad = await metadata_controller.handle_imageproxy(
446        _fake_request(f"/imageproxy/extra/{image_id}")
447    )
448    assert bad.status == 400
449    # trailing extra segment likewise
450    bad = await metadata_controller.handle_imageproxy(
451        _fake_request(f"/imageproxy/{image_id}/extra")
452    )
453    assert bad.status == 400
454    # double slash after the prefix must not be accepted either
455    bad = await metadata_controller.handle_imageproxy(_fake_request(f"/imageproxy//{image_id}"))
456    assert bad.status == 400
457
458
459async def test_handle_imageproxy_rejects_unsupported_size(
460    metadata_controller: MetaDataController,
461) -> None:
462    """A size outside the allowed set is rejected with a non-empty reason body."""
463    image_id = metadata_controller.compute_image_id("filesystem", "/local/cover.jpg")
464    request = MagicMock()
465    request.path = f"/imageproxy/{image_id}"
466    request.query = {"size": "300"}
467    bad = await metadata_controller.handle_imageproxy(request)
468    assert bad.status == 400
469    assert bad.text
470
471
472def test_is_svg_data() -> None:
473    """SVG bytes are detected by content, regardless of file extension."""
474    # plain root element
475    assert is_svg_data(b'<svg xmlns="http://www.w3.org/2000/svg"></svg>')
476    # XML declaration before the root element
477    assert is_svg_data(b'<?xml version="1.0"?>\n<svg viewBox="0 0 10 10"></svg>')
478    # leading whitespace and a comment before the root element
479    assert is_svg_data(b"  \n<!-- a logo -->\n<svg></svg>")
480    # doctype before the root element
481    assert is_svg_data(b'<!DOCTYPE svg PUBLIC "-//W3C//DTD SVG 1.1//EN">\n<svg></svg>')
482    # raster formats and arbitrary data are not SVG
483    assert not is_svg_data(b"\xff\xd8\xff\xe0JFIF")  # jpeg magic
484    assert not is_svg_data(b"\x89PNG\r\n\x1a\n")  # png magic
485    assert not is_svg_data(b"")
486    assert not is_svg_data(b"<html><body>not svg</body></html>")
487    # an XML document that never declares an <svg> element is not SVG
488    assert not is_svg_data(b'<?xml version="1.0"?><rss></rss>')
489
490
491def test_detect_image_content_format() -> None:
492    """Image format is sniffed from leading magic bytes (png/jpg/svg)."""
493    assert detect_image_content_format(b"\x89PNG\r\n\x1a\n....") == "png"
494    assert detect_image_content_format(b"\xff\xd8\xff\xe0JFIF") == "jpg"
495    assert detect_image_content_format(b'<svg xmlns="http://www.w3.org/2000/svg"></svg>') == "svg"
496    assert detect_image_content_format(b"") is None
497    assert detect_image_content_format(b"GIF89a") is None
498    # every sniffed value must map to a known content type
499    for fmt in ("png", "jpg", "svg"):
500        assert fmt in _IMAGEPROXY_CONTENT_TYPES
501
502
503def test_thumb_cache_filename_separates_flatten_variants() -> None:
504    """Flattened (player-compat) and transparency-preserving variants differ."""
505    thumb_hash = "a" * 64
506    plain = _thumb_cache_filename(thumb_hash, 512, "jpeg", flatten_transparency=False)
507    flat = _thumb_cache_filename(thumb_hash, 512, "jpeg", flatten_transparency=True)
508    assert plain == f"{thumb_hash}_512_v{_THUMB_CACHE_VERSION}.jpg"
509    assert flat == f"{thumb_hash}_512_v{_THUMB_CACHE_VERSION}_flat.jpg"
510    assert plain != flat
511    # png requests keep their own extension/bucket
512    png_name = _thumb_cache_filename(thumb_hash, 0, "png")
513    assert png_name == f"{thumb_hash}_0_v{_THUMB_CACHE_VERSION}.png"
514
515
516def test_has_alpha() -> None:
517    """Only images that actually use transparency are detected."""
518    assert _has_alpha(Image.new("RGBA", (4, 4), (0, 0, 0, 0)))
519    assert _has_alpha(Image.new("LA", (4, 4)))
520    palette_img = Image.new("P", (4, 4))
521    palette_img.info["transparency"] = 0
522    assert _has_alpha(palette_img)
523    # a fully-opaque alpha channel does not count as transparent
524    assert not _has_alpha(Image.new("RGBA", (4, 4), (1, 2, 3, 255)))
525    # opaque modes have no alpha
526    assert not _has_alpha(Image.new("RGB", (4, 4), (10, 20, 30)))
527    assert not _has_alpha(Image.new("L", (4, 4)))
528    assert not _has_alpha(Image.new("P", (4, 4)))
529
530
531def test_thumb_cache_filename_includes_cache_version() -> None:
532    """The version is in the filename so pre-upgrade thumbnails can't be reused."""
533    thumb_hash = "a" * 64
534    name = _thumb_cache_filename(thumb_hash, 512, "jpeg")
535    assert f"_v{_THUMB_CACHE_VERSION}" in name
536    # the unversioned (pre-PR) filename must not collide with the current one
537    assert name != f"{thumb_hash}_512.jpg"
538    # the sanitizer must accept the versioned shape it now produces
539    assert _THUMB_FILENAME_RE.fullmatch(name)
540    assert _THUMB_FILENAME_RE.fullmatch(
541        _thumb_cache_filename(thumb_hash, 0, "png", flatten_transparency=True)
542    )
543
544
545async def test_serve_thumbnail_sets_csp_for_svg(
546    metadata_controller: MetaDataController, monkeypatch: pytest.MonkeyPatch
547) -> None:
548    """SVG responses carry a script-blocking CSP; raster responses do not."""
549    served: dict[str, tuple[bytes, str]] = {"value": (b"<svg></svg>", "svg")}
550
551    async def _fake_resolve(*_args: object, **_kwargs: object) -> tuple[bytes, str]:
552        return served["value"]
553
554    monkeypatch.setattr(metadata_controller, "_resolve_thumbnail", _fake_resolve)
555
556    svg_resp = await metadata_controller._serve_thumbnail("p", "builtin", 0, "svg")
557    assert svg_resp.headers["Content-Security-Policy"] == (
558        "default-src 'none'; style-src 'unsafe-inline'; sandbox"
559    )
560    assert svg_resp.headers["X-Content-Type-Options"] == "nosniff"
561
562    served["value"] = (b"\xff\xd8\xff", "jpg")
563    jpg_resp = await metadata_controller._serve_thumbnail("p", "builtin", 256, "jpeg")
564    assert "Content-Security-Policy" not in jpg_resp.headers
565    assert "X-Content-Type-Options" not in jpg_resp.headers
566
567
568async def test_invalidate_image_cache_end_to_end(
569    metadata_controller: MetaDataController, tmp_path: Any, monkeypatch: pytest.MonkeyPatch
570) -> None:
571    """
572    Invalidation drops thumbs, source bytes and palette; the next request re-fetches.
573
574    This is the retag scenario: a local file's artwork is replaced while its
575    (provider, path) identity stays the same, so every derived artifact keyed
576    on that identity must be regenerated.
577    """
578    image_path = str(tmp_path / "cover.png")
579    Image.new("RGB", (300, 300), (200, 30, 30)).save(image_path, "PNG")
580    mass = metadata_controller.mass
581    fetches = 0
582    real_fetch = images_helper._fetch_source_image
583
584    async def counting_fetch(*args: Any, **kwargs: Any) -> tuple[bytes, bool]:
585        nonlocal fetches
586        fetches += 1
587        return await real_fetch(*args, **kwargs)
588
589    monkeypatch.setattr(images_helper, "_fetch_source_image", counting_fetch)
590
591    # derive two thumb variants and a palette from one single source fetch
592    await get_image_thumb(mass, image_path, 80, "builtin")
593    await get_image_thumb(mass, image_path, 256, "builtin")
594    palette = await get_palette(mass, image_path, "builtin")
595    assert palette is not None
596    assert palette.primary is not None
597    assert fetches == 1
598
599    thumb_hash = create_thumb_hash("builtin", image_path)
600    thumb_dir = Path(mass.cache_path, "thumbnails")
601    assert [f for f in thumb_dir.iterdir() if f.name.startswith(thumb_hash)]
602    assert await mass.cache.get(thumb_hash, provider="palette") is not None
603
604    await metadata_controller.invalidate_image_cache("builtin", image_path)
605
606    assert not [f for f in thumb_dir.iterdir() if f.name.startswith(thumb_hash)]
607    assert thumb_hash not in images_helper._source_memory_cache.entries
608    assert not any(key.startswith(f"{thumb_hash}_") for key in images_helper._thumb_memory_cache)
609    assert await mass.cache.get(thumb_hash, provider="palette") is None
610
611    # replace the artwork on disk: the next request serves the new content
612    Image.new("RGB", (300, 300), (30, 30, 200)).save(image_path, "PNG")
613    new_palette = await get_palette(mass, image_path, "builtin")
614    assert new_palette is not None
615    assert new_palette.primary != palette.primary
616
617
618def test_player_image_url_forces_jpeg_on_imageproxy_urls() -> None:
619    """Player-bound imageproxy URLs move to the streams server and force fmt=jpeg."""
620    mass = MagicMock()
621    mass.webserver.base_url = "http://192.168.1.2:8095"
622    mass.streams.base_url = "http://192.168.1.2:8097"
623    image_id = "a" * 64
624
625    url = f"http://192.168.1.2:8095/imageproxy/{image_id}?size=512&fmt=png"
626    result = player_image_url(mass, url)
627    assert result == f"http://192.168.1.2:8097/imageproxy/{image_id}?size=512&fmt=jpeg"
628
629    # non-imageproxy urls (e.g. remote radio artwork) pass through unchanged
630    remote = "https://example.com/art.png"
631    assert player_image_url(mass, remote) == remote
632    assert player_image_url(mass, None) is None
633