/
/
1"""Tests for the lightweight local metadata-file (NFO/image) change-detection cache."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import ANY, AsyncMock, MagicMock, patch
7
8import pytest
9from music_assistant_models.media_items import Artist, ProviderMapping, Track, UniqueList
10
11from music_assistant.providers.filesystem_local import LocalFileSystemProvider
12from music_assistant.providers.filesystem_local.constants import (
13 CACHE_CATEGORY_ALBUM_INFO,
14 CACHE_CATEGORY_ARTIST_INFO,
15 CACHE_CATEGORY_FOLDER_IMAGES,
16 CACHE_CATEGORY_METADATA_FILE,
17)
18from music_assistant.providers.filesystem_local.cue import make_cue_track_id
19from music_assistant.providers.filesystem_local.helpers import FileSystemItem
20
21INSTANCE_ID = "filesystem_local--test"
22
23
24def _provider() -> Any:
25 """Create a bare provider with a mocked cache."""
26 with patch.object(LocalFileSystemProvider, "__init__", lambda *_a, **_kw: None):
27 provider = LocalFileSystemProvider.__new__(LocalFileSystemProvider)
28 provider.logger = MagicMock()
29 provider.mass = MagicMock()
30 provider.config = MagicMock(instance_id=INSTANCE_ID)
31 provider.media_content_type = "music"
32 provider._sync_tracks = True
33 provider.cache = MagicMock()
34 return provider
35
36
37def _item(relative_path: str, checksum: str = "1") -> FileSystemItem:
38 """Build a minimal FileSystemItem, its checksum doubling as the metadata token."""
39 return FileSystemItem(
40 filename=relative_path.rsplit("/", 1)[-1],
41 relative_path=relative_path,
42 absolute_path=f"/media/{relative_path}",
43 is_dir=False,
44 checksum=checksum,
45 )
46
47
48# --- _queue_changed_metadata_files (walk-time change detection) ------------
49
50
51async def test_unchanged_metadata_file_is_ignored() -> None:
52 """A metadata file whose token still matches its cache entry queues nothing."""
53 provider = _provider()
54 meta = _item("Artist/Album/album.nfo", checksum="1")
55 provider.cache.get_all = AsyncMock(
56 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/t1.mp3"}}
57 )
58 items_to_process: list[tuple[FileSystemItem, str | None]] = []
59 force_refresh_tracks: set[str] = set()
60
61 await provider._queue_changed_metadata_files(
62 [meta], {}, {}, items_to_process, force_refresh_tracks
63 )
64
65 assert items_to_process == []
66
67
68async def test_registrations_are_bulk_loaded_once() -> None:
69 """Many metadata files trigger a single bulk cache load, not one lookup per file."""
70 provider = _provider()
71 provider.cache.get_all = AsyncMock(return_value={})
72 provider.cache.get = AsyncMock(side_effect=AssertionError("should not be called per-file"))
73 metadata_files = [_item(f"Artist/Album{i}/album.nfo") for i in range(25)]
74 items_to_process: list[tuple[FileSystemItem, str | None]] = []
75 force_refresh_tracks: set[str] = set()
76
77 await provider._queue_changed_metadata_files(
78 metadata_files, {}, {}, items_to_process, force_refresh_tracks
79 )
80
81 provider.cache.get_all.assert_awaited_once_with(
82 provider=INSTANCE_ID, category=CACHE_CATEGORY_METADATA_FILE
83 )
84 provider.cache.get.assert_not_called()
85
86
87async def test_changed_nfo_queues_representative_track() -> None:
88 """A changed NFO's registered representative track is queued for reparsing."""
89 provider = _provider()
90 meta = _item("Artist/Album/album.nfo", checksum="2")
91 provider.cache.get_all = AsyncMock(
92 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/t1.mp3"}}
93 )
94 track_item = _item("Artist/Album/t1.mp3")
95 provider.resolve = AsyncMock(return_value=track_item)
96 items_to_process: list[tuple[FileSystemItem, str | None]] = []
97 force_refresh_tracks: set[str] = set()
98
99 await provider._queue_changed_metadata_files(
100 [meta], {"Artist/Album/t1.mp3": "abc"}, {}, items_to_process, force_refresh_tracks
101 )
102
103 assert items_to_process == [(track_item, "abc")]
104 # its reparse must bypass this provider's short-lived album/artist caches, or an
105 # unrelated concurrent parse of the same folder could hand back the pre-change data
106 assert force_refresh_tracks == {"Artist/Album/t1.mp3"}
107
108
109async def test_changed_image_queues_representative_track_and_invalidates_it() -> None:
110 """A changed recognized folder image queues its representative and its own image cache."""
111 provider = _provider()
112 meta = _item("Artist/Album/folder.jpg", checksum="2")
113 provider.cache.get_all = AsyncMock(
114 return_value={"Artist/Album/folder.jpg": {"token": "1", "track": "Artist/Album/t1.mp3"}}
115 )
116 track_item = _item("Artist/Album/t1.mp3")
117 provider.resolve = AsyncMock(return_value=track_item)
118 provider.mass.metadata.invalidate_image_cache = AsyncMock()
119 items_to_process: list[tuple[FileSystemItem, str | None]] = []
120 force_refresh_tracks: set[str] = set()
121
122 await provider._queue_changed_metadata_files(
123 [meta], {}, {}, items_to_process, force_refresh_tracks
124 )
125
126 assert items_to_process == [(track_item, None)]
127 # the image itself keeps its (provider, path) identity, so its own cached thumbnail/source
128 # bytes must be invalidated directly: invalidating only the representative track is not
129 # enough since the track's path is never the image's path
130 provider.mass.metadata.invalidate_image_cache.assert_awaited_once_with(
131 INSTANCE_ID, "Artist/Album/folder.jpg"
132 )
133
134
135async def test_changed_nfo_does_not_invalidate_image_cache() -> None:
136 """A changed NFO (not an image) never triggers an image cache invalidation."""
137 provider = _provider()
138 meta = _item("Artist/Album/album.nfo", checksum="2")
139 provider.cache.get_all = AsyncMock(
140 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/t1.mp3"}}
141 )
142 provider.resolve = AsyncMock(return_value=_item("Artist/Album/t1.mp3"))
143 provider.mass.metadata.invalidate_image_cache = AsyncMock()
144 items_to_process: list[tuple[FileSystemItem, str | None]] = []
145 force_refresh_tracks: set[str] = set()
146
147 await provider._queue_changed_metadata_files(
148 [meta], {}, {}, items_to_process, force_refresh_tracks
149 )
150
151 provider.mass.metadata.invalidate_image_cache.assert_not_awaited()
152
153
154async def test_changed_metadata_file_for_cue_album_uses_cue_checksum_for_overwrite() -> None:
155 """
156 A CUE sheet queued as a representative gets its previous checksum from CUE tracking.
157
158 A CUE sheet's own path is never a key in `file_checksums` (only its synthetic per-track
159 ids are), so the CUE-specific checksum map must be consulted instead; otherwise the
160 reparse would look like a brand new import and skip overwriting the existing album.
161 """
162 provider = _provider()
163 meta = _item("Artist/Album/album.nfo", checksum="2")
164 provider.cache.get_all = AsyncMock(
165 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/album.cue"}}
166 )
167 cue_item = _item("Artist/Album/album.cue")
168 provider.resolve = AsyncMock(return_value=cue_item)
169 items_to_process: list[tuple[FileSystemItem, str | None]] = []
170 force_refresh_tracks: set[str] = set()
171
172 await provider._queue_changed_metadata_files(
173 [meta],
174 {}, # file_checksums: no direct entry for a CUE sheet's own path
175 {"Artist/Album/album.cue": {"cksum-a", "cksum-b"}},
176 items_to_process,
177 force_refresh_tracks,
178 )
179
180 assert items_to_process == [(cue_item, "cksum-a")] # min() of the tracked set
181
182
183async def test_changed_metadata_file_for_cue_artist_queues_one_cue_item() -> None:
184 """
185 An artist.nfo/image registered against a CUE representative queues the CUE item itself.
186
187 Reprocessing the CUE sheet re-runs CUE track parsing, which rebuilds every track (and
188 thus every artist/album) it describes - so a single queued CUE item is enough, and no
189 synthetic per-track id is ever resolved directly.
190 """
191 provider = _provider()
192 nfo = _item("Artist/artist.nfo", checksum="2")
193 img = _item("Artist/artist.jpg", checksum="9")
194 provider.cache.get_all = AsyncMock(
195 return_value={
196 "Artist/artist.nfo": {"token": "1", "track": "Artist/Album/album.cue"},
197 "Artist/artist.jpg": {"token": "8", "track": "Artist/Album/album.cue"},
198 }
199 )
200 cue_item = _item("Artist/Album/album.cue")
201 provider.resolve = AsyncMock(return_value=cue_item)
202 provider.mass.metadata.invalidate_image_cache = AsyncMock()
203 items_to_process: list[tuple[FileSystemItem, str | None]] = []
204 force_refresh_tracks: set[str] = set()
205
206 await provider._queue_changed_metadata_files(
207 [nfo, img],
208 {},
209 {"Artist/Album/album.cue": {"cksum-a"}},
210 items_to_process,
211 force_refresh_tracks,
212 )
213
214 assert items_to_process == [(cue_item, "cksum-a")] # queued once, deduped across both files
215 assert force_refresh_tracks == {"Artist/Album/album.cue"}
216
217
218async def test_register_metadata_file_stores_cue_path_as_representative() -> None:
219 """Registering against a CUE-derived representative stores the CUE path, not a synthetic id."""
220 provider = _provider()
221 provider.cache.set = AsyncMock()
222 meta = _item("Artist/artist.nfo", checksum="7")
223
224 await provider._register_metadata_file(meta, "Artist/Album/album.cue")
225
226 provider.cache.set.assert_awaited_once_with(
227 key="Artist/artist.nfo",
228 data={"token": "7", "track": "Artist/Album/album.cue"},
229 provider=INSTANCE_ID,
230 category=CACHE_CATEGORY_METADATA_FILE,
231 expiration=ANY,
232 persistent=True,
233 )
234
235
236async def test_missing_cue_representative_track_defers() -> None:
237 """A CUE representative that no longer resolves is skipped, retried on a later sync."""
238 provider = _provider()
239 meta = _item("Artist/artist.nfo", checksum="2")
240 provider.cache.get_all = AsyncMock(
241 return_value={"Artist/artist.nfo": {"token": "1", "track": "Artist/Album/gone.cue"}}
242 )
243 provider.resolve = AsyncMock(side_effect=FileNotFoundError())
244 provider.cache.set = AsyncMock()
245 items_to_process: list[tuple[FileSystemItem, str | None]] = []
246 force_refresh_tracks: set[str] = set()
247
248 await provider._queue_changed_metadata_files(
249 [meta], {}, {}, items_to_process, force_refresh_tracks
250 )
251
252 assert items_to_process == []
253 assert provider.cache.set.await_args_list == [] # old token kept, so a later sync retries
254
255
256async def test_two_changed_metadata_files_dedupe_to_one_track() -> None:
257 """Two changed metadata files sharing a representative queue it only once."""
258 provider = _provider()
259 nfo = _item("Artist/Album/album.nfo", checksum="2")
260 img = _item("Artist/Album/folder.jpg", checksum="9")
261 provider.cache.get_all = AsyncMock(
262 return_value={
263 "Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/t1.mp3"},
264 "Artist/Album/folder.jpg": {"token": "8", "track": "Artist/Album/t1.mp3"},
265 }
266 )
267 track_item = _item("Artist/Album/t1.mp3")
268 provider.resolve = AsyncMock(return_value=track_item)
269 provider.mass.metadata.invalidate_image_cache = AsyncMock()
270 items_to_process: list[tuple[FileSystemItem, str | None]] = []
271 force_refresh_tracks: set[str] = set()
272
273 await provider._queue_changed_metadata_files(
274 [nfo, img], {}, {}, items_to_process, force_refresh_tracks
275 )
276
277 assert len(items_to_process) == 1
278 assert items_to_process[0][0] is track_item
279
280
281async def test_track_already_changed_is_not_duplicated() -> None:
282 """A representative already queued (its own content changed) is not queued twice."""
283 provider = _provider()
284 meta = _item("Artist/Album/album.nfo", checksum="2")
285 provider.cache.get_all = AsyncMock(
286 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/t1.mp3"}}
287 )
288 provider.resolve = AsyncMock()
289 existing = (_item("Artist/Album/t1.mp3"), "old")
290 items_to_process: list[tuple[FileSystemItem, str | None]] = [existing]
291 force_refresh_tracks: set[str] = set()
292
293 await provider._queue_changed_metadata_files(
294 [meta], {}, {}, items_to_process, force_refresh_tracks
295 )
296
297 assert items_to_process == [existing]
298 provider.resolve.assert_not_awaited()
299 # the track's own change already queued it, but the NFO change happening in the same sync
300 # must still force a fresh reparse - otherwise a concurrent parse of the same folder could
301 # serve stale cached data and silently miss this simultaneous metadata edit
302 assert force_refresh_tracks == {"Artist/Album/t1.mp3"}
303
304
305async def test_cache_miss_is_ignored() -> None:
306 """A metadata file with no cache entry (new/untracked) queues nothing."""
307 provider = _provider()
308 meta = _item("Artist/Album/album.nfo")
309 provider.cache.get_all = AsyncMock(return_value={})
310 items_to_process: list[tuple[FileSystemItem, str | None]] = []
311 force_refresh_tracks: set[str] = set()
312
313 await provider._queue_changed_metadata_files(
314 [meta], {}, {}, items_to_process, force_refresh_tracks
315 )
316
317 assert items_to_process == []
318
319
320async def test_missing_representative_track_defers() -> None:
321 """A representative track that no longer resolves is skipped, not raised or written."""
322 provider = _provider()
323 meta = _item("Artist/Album/album.nfo", checksum="2")
324 provider.cache.get_all = AsyncMock(
325 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/gone.mp3"}}
326 )
327 provider.resolve = AsyncMock(side_effect=FileNotFoundError())
328 provider.cache.set = AsyncMock()
329 items_to_process: list[tuple[FileSystemItem, str | None]] = []
330 force_refresh_tracks: set[str] = set()
331
332 await provider._queue_changed_metadata_files(
333 [meta], {}, {}, items_to_process, force_refresh_tracks
334 )
335
336 assert items_to_process == []
337 provider.cache.set.assert_not_awaited() # old token kept, so a later sync retries
338 # nothing was actually queued, so the batch has no reason to drop its short-lived caches
339 assert force_refresh_tracks == set()
340
341
342# --- _classify_scan_item (walk routing, shared by local/WebDAV/cloud) ------
343
344
345def test_classify_scan_item_routes_metadata_file_without_recording_it() -> None:
346 """A recognized metadata file is collected separately and never treated as media."""
347 provider = _provider()
348 item = _item("Artist/Album/album.nfo")
349 items_to_process: list[tuple[FileSystemItem, str | None]] = []
350 cur_filenames: set[str] = set()
351 metadata_files: list[FileSystemItem] = []
352
353 provider._classify_scan_item(
354 item,
355 file_checksums={},
356 cue_file_checksums={},
357 cur_filenames=cur_filenames,
358 items_to_process=items_to_process,
359 unchanged_cue_items=[],
360 cue_stems=set(),
361 ignore_album_playlists=False,
362 metadata_files=metadata_files,
363 )
364
365 assert metadata_files == [item]
366 assert items_to_process == []
367 assert cur_filenames == set() # never present/absent-tracked, so never deleted either
368
369
370def test_classify_scan_item_ignores_unrecognized_metadata_extension_files() -> None:
371 """
372 An unrecognized nfo/image (walked for its extension, not a recognized filename).
373
374 Must stay fully invisible to the scan - not routed to metadata_files (it isn't a
375 recognized metadata file) and not counted in cur_filenames either, or a stray file
376 like this on a wrong/empty mount would satisfy the "not empty" check and silently
377 bypass the safeguard against deleting an entire previously-synced library.
378 """
379 provider = _provider()
380 unrecognized_image = _item("Artist/Album/booklet.jpg")
381 unrecognized_nfo = _item("Artist/Album/random.nfo")
382 for item in (unrecognized_image, unrecognized_nfo):
383 items_to_process: list[tuple[FileSystemItem, str | None]] = []
384 cur_filenames: set[str] = set()
385 metadata_files: list[FileSystemItem] = []
386
387 provider._classify_scan_item(
388 item,
389 file_checksums={},
390 cue_file_checksums={},
391 cur_filenames=cur_filenames,
392 items_to_process=items_to_process,
393 unchanged_cue_items=[],
394 cue_stems=set(),
395 ignore_album_playlists=False,
396 metadata_files=metadata_files,
397 )
398
399 assert metadata_files == []
400 assert items_to_process == []
401 assert cur_filenames == set()
402
403
404def test_track_classification_uses_checksum_not_metadata_token() -> None:
405 """A track's own change detection is driven only by checksum, imported-media compatible."""
406 provider = _provider()
407 item = FileSystemItem(
408 filename="track.mp3",
409 relative_path="Artist/Album/track.mp3",
410 absolute_path="/media/Artist/Album/track.mp3",
411 is_dir=False,
412 checksum="100",
413 metadata_token="999999999999", # unrelated high-precision value, must not matter here
414 )
415 items_to_process: list[tuple[FileSystemItem, str | None]] = []
416 cur_filenames: set[str] = set()
417
418 provider._classify_scan_item(
419 item,
420 file_checksums={"Artist/Album/track.mp3": "100"},
421 cue_file_checksums={},
422 cur_filenames=cur_filenames,
423 items_to_process=items_to_process,
424 unchanged_cue_items=[],
425 cue_stems=set(),
426 ignore_album_playlists=False,
427 metadata_files=[],
428 )
429
430 assert items_to_process == []
431 assert "Artist/Album/track.mp3" in cur_filenames
432
433
434# --- _register_metadata_file / _parse_artist integration -------------------
435
436
437async def test_register_metadata_file_without_representative_is_a_no_op() -> None:
438 """A metadata file read outside any track context (no representative) is never cached."""
439 provider = _provider()
440 provider.cache.set = AsyncMock()
441
442 await provider._register_metadata_file(_item("Artist/artist.nfo"), None)
443
444 provider.cache.set.assert_not_awaited()
445
446
447async def test_parse_artist_registers_metadata_file_after_successful_read() -> None:
448 """Reading artist.nfo during parsing registers its token and representative track."""
449 provider = _provider()
450 provider.manifest = MagicMock(domain="filesystem_local")
451 provider.exists = AsyncMock(return_value=True)
452 provider._read_file = AsyncMock(return_value=b"<artist><title>Name</title></artist>")
453 provider._get_local_images = AsyncMock(return_value=UniqueList())
454 provider.resolve = AsyncMock(return_value=_item("Artist/artist.nfo", checksum="42"))
455 provider.cache.get = AsyncMock(return_value=None)
456 provider.cache.set = AsyncMock()
457
458 await provider._parse_artist(
459 "Name", artist_path="Artist", representative_track="Artist/Album/t1.mp3"
460 )
461
462 meta_calls = [
463 call
464 for call in provider.cache.set.await_args_list
465 if call.kwargs.get("category") == CACHE_CATEGORY_METADATA_FILE
466 ]
467 assert len(meta_calls) == 1
468 assert meta_calls[0].kwargs["key"] == "Artist/artist.nfo"
469 assert meta_calls[0].kwargs["data"] == {"token": "42", "track": "Artist/Album/t1.mp3"}
470
471
472async def test_parse_artist_does_not_register_when_nfo_read_fails() -> None:
473 """A transient read failure while parsing artist.nfo never advances its cached token."""
474 provider = _provider()
475 provider.manifest = MagicMock(domain="filesystem_local")
476 provider.exists = AsyncMock(return_value=True)
477 provider.resolve = AsyncMock(return_value=_item("Artist/artist.nfo"))
478 provider._read_file = AsyncMock(side_effect=OSError("network blip"))
479 provider.cache.get = AsyncMock(return_value=None)
480 provider.cache.set = AsyncMock()
481
482 with pytest.raises(OSError, match="network blip"):
483 await provider._parse_artist(
484 "Name", artist_path="Artist", representative_track="Artist/Album/t1.mp3"
485 )
486
487 provider.cache.set.assert_not_awaited()
488
489
490async def test_parse_artist_does_not_register_on_malformed_nfo() -> None:
491 """
492 Malformed (but readable) artist.nfo XML is warned about, not registered as handled.
493
494 Registering here would advance the token and make the malformed edit look already
495 processed, so the same broken file would never be retried once it is eventually fixed.
496 """
497 provider = _provider()
498 provider.manifest = MagicMock(domain="filesystem_local")
499 provider.exists = AsyncMock(return_value=True)
500 provider.resolve = AsyncMock(return_value=_item("Artist/artist.nfo"))
501 provider._read_file = AsyncMock(return_value=b"not xml at all <<<")
502 provider._get_local_images = AsyncMock(return_value=UniqueList())
503 provider.cache.get = AsyncMock(return_value=None)
504 provider.cache.set = AsyncMock()
505
506 artist = await provider._parse_artist(
507 "Name", artist_path="Artist", representative_track="Artist/Album/t1.mp3"
508 )
509
510 assert artist is not None # the malformed NFO is only warned about, not fatal
511 meta_calls = [
512 call
513 for call in provider.cache.set.await_args_list
514 if call.kwargs.get("category") == CACHE_CATEGORY_METADATA_FILE
515 ]
516 assert meta_calls == []
517
518
519async def test_parse_album_does_not_register_on_malformed_nfo() -> None:
520 """Malformed (but readable) album.nfo XML is warned about, not registered as handled."""
521 provider = _provider()
522 provider.manifest = MagicMock(domain="filesystem_local")
523 provider.exists = AsyncMock(return_value=True)
524 provider.resolve = AsyncMock(return_value=_item("Artist/Album/album.nfo"))
525 provider._read_file = AsyncMock(return_value=b"not xml at all <<<")
526 provider._get_local_images = AsyncMock(return_value=UniqueList())
527 provider.cache.get = AsyncMock(return_value=None)
528 provider.cache.set = AsyncMock()
529 provider._resolve_artists_with_mbids = AsyncMock(return_value=[])
530 provider.config.get_value = MagicMock(return_value="various_artists")
531
532 tags = MagicMock(
533 album="My Album",
534 album_artists=[],
535 album_sort=None,
536 barcode=None,
537 musicbrainz_albumid=None,
538 musicbrainz_releasegroupid=None,
539 year=None,
540 album_type=None,
541 filename="track.mp3",
542 )
543 album = await provider._parse_album(track_path="Artist/Album/t1.mp3", track_tags=tags)
544
545 assert album is not None
546 meta_calls = [
547 call
548 for call in provider.cache.set.await_args_list
549 if call.kwargs.get("category") == CACHE_CATEGORY_METADATA_FILE
550 ]
551 assert meta_calls == []
552
553
554# --- force_refresh_tracks signals a stale-cache-clearing need -------------
555
556
557async def test_queue_changed_metadata_files_marks_representative_for_force_refresh() -> None:
558 """A representative queued from a metadata-file change is marked for cache-bypassing."""
559 provider = _provider()
560 meta = _item("Artist/Album/album.nfo", checksum="2")
561 provider.cache.get_all = AsyncMock(
562 return_value={"Artist/Album/album.nfo": {"token": "1", "track": "Artist/Album/t1.mp3"}}
563 )
564 track_item = _item("Artist/Album/t1.mp3")
565 provider.resolve = AsyncMock(return_value=track_item)
566 items_to_process: list[tuple[FileSystemItem, str | None]] = []
567 force_refresh_tracks: set[str] = set()
568
569 await provider._queue_changed_metadata_files(
570 [meta], {}, {}, items_to_process, force_refresh_tracks
571 )
572
573 assert force_refresh_tracks == {"Artist/Album/t1.mp3"}
574
575
576async def test_drop_stale_album_artist_caches_clears_the_three_categories() -> None:
577 """The stale-cache drop targets exactly album/artist/folder-image, scoped to this provider."""
578 provider = _provider()
579 provider.cache.delete = AsyncMock()
580
581 await provider._drop_stale_album_artist_caches()
582
583 calls = provider.cache.delete.await_args_list
584 categories = {call.kwargs["category"] for call in calls}
585 assert categories == {
586 CACHE_CATEGORY_ALBUM_INFO,
587 CACHE_CATEGORY_ARTIST_INFO,
588 CACHE_CATEGORY_FOLDER_IMAGES,
589 }
590 assert all(call.kwargs["key"] is None for call in calls)
591 assert all(call.kwargs["provider"] == INSTANCE_ID for call in calls)
592
593
594# --- get_artist registers a representative for manual "Refresh item" ------
595
596
597async def test_resolve_artist_representative_track_uses_own_provider_mapping() -> None:
598 """The artist's own instance-local track path is picked as its representative."""
599 provider = _provider()
600 artist = Artist(
601 item_id="42",
602 provider="library",
603 name="Test Artist",
604 provider_mappings=set(),
605 )
606 other_track = Track(
607 item_id="other",
608 provider="library",
609 name="Other Provider Track",
610 provider_mappings={
611 ProviderMapping(
612 item_id="Other/track.flac", provider_domain="qobuz", provider_instance="qobuz"
613 )
614 },
615 )
616 own_track = Track(
617 item_id="mine",
618 provider="library",
619 name="Local Track",
620 provider_mappings={
621 ProviderMapping(
622 item_id="Artist/Album/t1.mp3",
623 provider_domain="filesystem_local",
624 provider_instance=INSTANCE_ID,
625 )
626 },
627 )
628 provider.mass.music.artists.get_library_artist_tracks = AsyncMock(
629 return_value=[other_track, own_track]
630 )
631
632 result = await provider._resolve_artist_representative_track(artist)
633
634 assert result == "Artist/Album/t1.mp3"
635 provider.mass.music.artists.get_library_artist_tracks.assert_awaited_once_with(
636 "42", provider_filter=INSTANCE_ID
637 )
638
639
640async def test_resolve_artist_representative_track_returns_none_without_tracks() -> None:
641 """An artist with no tracks under this provider (album-only) yields no representative."""
642 provider = _provider()
643 artist = Artist(item_id="42", provider="library", name="Test Artist", provider_mappings=set())
644 provider.mass.music.artists.get_library_artist_tracks = AsyncMock(return_value=[])
645
646 assert await provider._resolve_artist_representative_track(artist) is None
647
648
649async def test_resolve_artist_representative_track_converts_cue_track_id_to_cue_path() -> None:
650 """
651 A CUE-derived track's synthetic id must resolve to its (real, resolvable) CUE path.
652
653 A CUE track's own provider mapping id is a synthetic "<cue path>::<track>" marker, not a
654 path on disk; returning it verbatim would make `_queue_changed_metadata_files` try to
655 resolve a nonexistent file and defer this artist's NFO/image changes forever.
656 """
657 provider = _provider()
658 artist = Artist(item_id="42", provider="library", name="Test Artist", provider_mappings=set())
659 cue_track = Track(
660 item_id="cue-track",
661 provider="library",
662 name="CUE Track",
663 provider_mappings={
664 ProviderMapping(
665 item_id=make_cue_track_id("Artist/Album/album.cue", 3),
666 provider_domain="filesystem_local",
667 provider_instance=INSTANCE_ID,
668 )
669 },
670 )
671 provider.mass.music.artists.get_library_artist_tracks = AsyncMock(return_value=[cue_track])
672
673 result = await provider._resolve_artist_representative_track(artist)
674
675 assert result == "Artist/Album/album.cue"
676
677
678async def test_get_artist_passes_representative_track_to_parse_artist() -> None:
679 """
680 Fetching a library artist threads a representative track into `_parse_artist`.
681
682 Without this, a manual "Refresh item" on an artist would read (and re-cache) its NFO/image
683 without ever registering them, leaving the documented manual-refresh fallback a no-op.
684 """
685 provider = _provider()
686 provider.manifest = MagicMock(domain="filesystem_local")
687 db_artist = Artist(
688 item_id="42",
689 provider="library",
690 name="Test Artist",
691 sort_name="Test Artist",
692 provider_mappings={
693 ProviderMapping(
694 item_id="Artist",
695 provider_domain="filesystem_local",
696 provider_instance=INSTANCE_ID,
697 url="Artist",
698 )
699 },
700 )
701 provider.mass.music.artists.get_library_item_by_prov_id = AsyncMock(return_value=db_artist)
702 provider.mass.music.artists.get_library_artist_tracks = AsyncMock(
703 return_value=[
704 Track(
705 item_id="mine",
706 provider="library",
707 name="Local Track",
708 provider_mappings={
709 ProviderMapping(
710 item_id="Artist/Album/t1.mp3",
711 provider_domain="filesystem_local",
712 provider_instance=INSTANCE_ID,
713 )
714 },
715 )
716 ]
717 )
718 provider.exists = AsyncMock(return_value=True)
719 provider._parse_artist = AsyncMock(return_value=db_artist)
720
721 await provider.get_artist("Artist")
722
723 provider._parse_artist.assert_awaited_once()
724 assert provider._parse_artist.await_args.kwargs["representative_track"] == (
725 "Artist/Album/t1.mp3"
726 )
727