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