/
/
/
1"""Tests for image-cache invalidation when a local file changes during sync."""
2
3from __future__ import annotations
4
5from unittest.mock import AsyncMock, MagicMock, patch
6
7from music_assistant.providers.filesystem_local import LocalFileSystemProvider
8
9
10def _create_provider() -> tuple[LocalFileSystemProvider, AsyncMock]:
11 """Create a bare LocalFileSystemProvider plus the invalidation mock it calls."""
12 with patch.object(LocalFileSystemProvider, "__init__", lambda *_a, **_kw: None):
13 provider = LocalFileSystemProvider.__new__(LocalFileSystemProvider)
14 provider.logger = MagicMock()
15 invalidate_mock = AsyncMock()
16 provider.mass = MagicMock()
17 provider.mass.metadata.invalidate_image_cache = invalidate_mock
18 provider.media_content_type = "music"
19 provider.config = MagicMock(instance_id="filesystem_local--test")
20 return provider, invalidate_mock
21
22
23def _file_item(relative_path: str) -> MagicMock:
24 """Build a minimal FileSystemItem stand-in for _process_item_async."""
25 item = MagicMock()
26 item.relative_path = relative_path
27 item.absolute_path = f"/music/{relative_path}"
28 # unhandled extension: the sync branches are skipped, only the shared
29 # invalidation logic at the top of _process_item_async runs
30 item.ext = "unhandled"
31 return item
32
33
34async def test_changed_file_invalidates_both_image_path_forms() -> None:
35 """A file with a previous checksum (= changed) busts both image path forms."""
36 provider, invalidate_mock = _create_provider()
37 item = _file_item("Artist/Album/track.mp3")
38 await provider._process_item_async(item, prev_checksum="12345")
39 assert invalidate_mock.await_count == 2
40 invalidate_mock.assert_any_await("filesystem_local--test", "Artist/Album/track.mp3")
41 invalidate_mock.assert_any_await("filesystem_local--test", "Artist/Album/track.mp3?cs=12345")
42
43
44async def test_new_file_does_not_invalidate() -> None:
45 """A file seen for the first time (no previous checksum) busts nothing."""
46 provider, invalidate_mock = _create_provider()
47 item = _file_item("Artist/Album/new-track.mp3")
48 await provider._process_item_async(item, prev_checksum=None)
49 invalidate_mock.assert_not_awaited()
50