/
/
/
1"""Integration tests for CUE sheet support in the filesystem provider."""
2
3from __future__ import annotations
4
5from pathlib import Path
6from typing import cast
7from unittest.mock import AsyncMock, MagicMock, patch
8
9import pytest
10from music_assistant_models.enums import ContentType, ExternalID
11from music_assistant_models.errors import InvalidDataError, MediaNotFoundError
12from music_assistant_models.media_items import Album, Artist, AudioFormat, Track
13from music_assistant_models.streamdetails import StreamDetails
14
15from music_assistant.constants import UNKNOWN_ARTIST
16from music_assistant.helpers.tags import AudioTags
17from music_assistant.providers.filesystem_local import LocalFileSystemProvider
18from music_assistant.providers.filesystem_local.cue import (
19 CUE_TRACK_ID_DELIMITER,
20 CueSheetHandler,
21 cue_metadata_checksum,
22 make_cue_track_id,
23 parse_cue_track_id,
24)
25from music_assistant.providers.filesystem_local.helpers import FileSystemItem
26
27SAMPLE_CUE = """\
28REM GENRE "Classic Rock"
29REM DATE 1995
30PERFORMER "Dire Straits"
31TITLE "Live at the BBC"
32FILE "album.flac" WAVE
33 TRACK 01 AUDIO
34 TITLE "Down to the Waterline"
35 PERFORMER "Dire Straits"
36 ISRC GBAMU7800001
37 INDEX 01 00:00:00
38 TRACK 02 AUDIO
39 TITLE "Six Blade Knife"
40 PERFORMER "Dire Straits"
41 ISRC GBAMU7800002
42 INDEX 01 04:10:40
43 TRACK 03 AUDIO
44 TITLE "Water of Love"
45 PERFORMER "Dire Straits"
46 ISRC GBAMU7800003
47 INDEX 01 07:58:05
48"""
49
50
51def _make_audio_tags(
52 duration: float = 900.0,
53 album: str | None = None,
54 albumartist: str | None = None,
55 genre: str | None = None,
56 disc: str | None = None,
57 has_cover_image: bool = False,
58 **extra_tags: str,
59) -> AudioTags:
60 """Build a minimal AudioTags object for tests."""
61 tag_dict: dict[str, str] = {}
62 if album is not None:
63 tag_dict["album"] = album
64 if albumartist is not None:
65 tag_dict["albumartist"] = albumartist
66 if genre is not None:
67 tag_dict["genre"] = genre
68 if disc is not None:
69 tag_dict["disc"] = disc
70 tag_dict.update(extra_tags)
71 return AudioTags(
72 raw={},
73 sample_rate=44100,
74 channels=2,
75 bits_per_sample=16,
76 format="flac",
77 bit_rate=1000,
78 duration=duration,
79 tags=tag_dict,
80 has_cover_image=has_cover_image,
81 filename="album.flac",
82 )
83
84
85def _make_provider(base_path: str = "/music") -> LocalFileSystemProvider:
86 """Build a LocalFileSystemProvider with dependencies mocked."""
87 with patch.object(LocalFileSystemProvider, "__init__", lambda *_a, **_kw: None):
88 provider = LocalFileSystemProvider.__new__(LocalFileSystemProvider)
89 provider.media_content_type = "music"
90 provider.base_path = base_path
91 # instance_id and domain are read-only properties sourced from config/manifest
92 provider.config = MagicMock(instance_id="filesystem_local--test")
93 provider.manifest = MagicMock(domain="filesystem_local")
94 provider.logger = MagicMock()
95 provider.mass = MagicMock()
96 # cache is used by load_cue_sheet; default to miss so tests exercise the parse path
97 provider.mass.cache.get = AsyncMock(return_value=None)
98 provider.mass.cache.set = AsyncMock(return_value=None)
99 provider.cache = MagicMock()
100 provider._sync_tracks = True
101 provider.sync_running = False
102 provider._sync_nfo_by_dir = {}
103 provider._sync_nfo_index_ready = False
104 provider._cue = CueSheetHandler(provider)
105 return provider
106
107
108def _stub_library_track(
109 provider: LocalFileSystemProvider, item_id: str, duration: int = 180
110) -> None:
111 """Configure the provider's mass to return a library track for item_id."""
112 prov_mapping = MagicMock(
113 item_id=item_id,
114 audio_format=AudioFormat(
115 content_type=ContentType.FLAC,
116 sample_rate=44100,
117 bit_depth=16,
118 channels=2,
119 bit_rate=1000,
120 ),
121 )
122 library_track = MagicMock(provider_mappings=[prov_mapping], duration=duration)
123 provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=library_track) # type: ignore[method-assign]
124
125
126def _make_cue_item(tmp_path: Path, cue_text: str, name: str = "album.cue") -> FileSystemItem:
127 """Write a CUE file under tmp_path and return a FileSystemItem for it."""
128 cue_file = tmp_path / name
129 cue_file.write_text(cue_text, encoding="utf-8")
130 return FileSystemItem(
131 filename=name,
132 relative_path=name,
133 absolute_path=str(cue_file),
134 is_dir=False,
135 checksum="1",
136 file_size=cue_file.stat().st_size,
137 created_at=1700000000,
138 )
139
140
141class TestCueTrackIdHelpers:
142 """Tests for CUE track id construction and parsing."""
143
144 def test_make_format(self) -> None:
145 """Make format."""
146 assert make_cue_track_id("album.cue", 3) == f"album.cue{CUE_TRACK_ID_DELIMITER}03"
147
148 def test_make_pads_single_digits(self) -> None:
149 """Make pads single digits."""
150 assert make_cue_track_id("a.cue", 1).endswith("01")
151
152 def test_make_handles_large_track_numbers(self) -> None:
153 """Make handles large track numbers."""
154 assert make_cue_track_id("a.cue", 123).endswith("123")
155
156 def test_parse_roundtrip(self) -> None:
157 """Parse roundtrip."""
158 for track_num in (1, 9, 10, 99):
159 item_id = make_cue_track_id("artist/album.cue", track_num)
160 parsed = parse_cue_track_id(item_id)
161 assert parsed == ("artist/album.cue", track_num)
162
163 def test_parse_non_cue_id_returns_none(self) -> None:
164 """Parse non cue id returns none."""
165 assert parse_cue_track_id("regular/path/track.flac") is None
166 assert parse_cue_track_id("") is None
167
168
169class TestReadCueFile:
170 """Tests for CUE file encoding handling."""
171
172 @pytest.mark.asyncio
173 async def test_reads_utf8(self, tmp_path: Path) -> None:
174 """Reads utf8."""
175 (tmp_path / "a.cue").write_text('TITLE "Café"\n', encoding="utf-8")
176 provider = _make_provider(base_path=str(tmp_path))
177 cue_item = _make_cue_item(tmp_path, 'TITLE "Café"\n', name="a.cue")
178 content = await provider._cue.read_cue_file(cue_item)
179 assert "Café" in content
180
181 @pytest.mark.asyncio
182 async def test_reads_utf8_bom(self, tmp_path: Path) -> None:
183 """Reads utf8 bom."""
184 (tmp_path / "a.cue").write_text('TITLE "Café"\n', encoding="utf-8-sig")
185 provider = _make_provider(base_path=str(tmp_path))
186 cue_item = FileSystemItem(
187 filename="a.cue",
188 relative_path="a.cue",
189 absolute_path=str(tmp_path / "a.cue"),
190 is_dir=False,
191 checksum="1",
192 file_size=(tmp_path / "a.cue").stat().st_size,
193 )
194 content = await provider._cue.read_cue_file(cue_item)
195 assert "Café" in content
196 assert not content.startswith("\ufeff")
197
198 @pytest.mark.asyncio
199 async def test_decodes_latin1_bytes(self, tmp_path: Path) -> None:
200 """Bytes that are not valid UTF-8 keep their accented characters."""
201 # 0xFC is "ü" in Latin-1 but invalid as a UTF-8 continuation byte
202 (tmp_path / "a.cue").write_bytes(b'TITLE "M\xfcller"\n')
203 provider = _make_provider(base_path=str(tmp_path))
204 cue_item = FileSystemItem(
205 filename="a.cue",
206 relative_path="a.cue",
207 absolute_path=str(tmp_path / "a.cue"),
208 is_dir=False,
209 checksum="1",
210 file_size=(tmp_path / "a.cue").stat().st_size,
211 )
212 content = await provider._cue.read_cue_file(cue_item)
213 assert content == 'TITLE "Müller"\n'
214
215 @pytest.mark.asyncio
216 async def test_reads_cyrillic_cue_sheet(self, tmp_path: Path) -> None:
217 """
218 A CUE sheet in the local ANSI codepage keeps its titles readable.
219
220 Russian rips ship their CUE sheets in cp1251, so the titles have to come
221 through as Cyrillic instead of replacement characters (support #6093).
222 """
223 cue = 'PERFORMER "ÐоÑÐ¾Ð»Ñ Ð¸ ШÑÑ"\nTITLE "Ðак в ÑÑаÑой Ñказке"\n'
224 (tmp_path / "a.cue").write_bytes(cue.encode("cp1251"))
225 provider = _make_provider(base_path=str(tmp_path))
226 cue_item = FileSystemItem(
227 filename="a.cue",
228 relative_path="a.cue",
229 absolute_path=str(tmp_path / "a.cue"),
230 is_dir=False,
231 checksum="1",
232 file_size=(tmp_path / "a.cue").stat().st_size,
233 )
234 content = await provider._cue.read_cue_file(cue_item)
235 assert content == cue
236
237
238class TestLoadCueSheet:
239 """Tests for parsed CUE sheet caching."""
240
241 @pytest.mark.asyncio
242 async def test_versions_cache_entries(self, tmp_path: Path) -> None:
243 """Parsed metadata uses a checksum that changes with CUE handling."""
244 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
245 provider = _make_provider(base_path=str(tmp_path))
246 cache_get = cast("AsyncMock", provider.mass.cache.get)
247 cache_set = cast("AsyncMock", provider.mass.cache.set)
248
249 await provider._cue.load_cue_sheet(cue_item)
250
251 expected_checksum = cue_metadata_checksum(cue_item.checksum)
252 get_call = cache_get.await_args
253 set_call = cache_set.await_args
254 assert get_call is not None
255 assert set_call is not None
256 assert get_call.kwargs["checksum"] == expected_checksum
257 assert set_call.kwargs["checksum"] == expected_checksum
258
259
260class TestFindCueAudioFile:
261 """Tests for audio file resolution from a CUE sheet."""
262
263 @pytest.mark.asyncio
264 async def test_matches_file_command(self, tmp_path: Path) -> None:
265 """Matches file command."""
266 (tmp_path / "album.flac").write_bytes(b"")
267 (tmp_path / "other.flac").write_bytes(b"")
268 cue_item = _make_cue_item(tmp_path, 'FILE "album.flac" WAVE\n')
269 provider = _make_provider(base_path=str(tmp_path))
270 cue_sheet = MagicMock(file_path="album.flac")
271 result = await provider._cue.find_audio_file(cue_item, cue_sheet)
272 assert result == "album.flac"
273
274 @pytest.mark.asyncio
275 async def test_same_stem_fallback(self, tmp_path: Path) -> None:
276 """Same stem fallback."""
277 (tmp_path / "album.flac").write_bytes(b"")
278 cue_item = _make_cue_item(tmp_path, "")
279 provider = _make_provider(base_path=str(tmp_path))
280 cue_sheet = MagicMock(file_path=None)
281 result = await provider._cue.find_audio_file(cue_item, cue_sheet)
282 assert result == "album.flac"
283
284 @pytest.mark.asyncio
285 async def test_returns_none_when_file_missing_and_stem_mismatch(self, tmp_path: Path) -> None:
286 """Returns None when neither FILE nor same-stem match locates the audio file."""
287 (tmp_path / "onlyone.flac").write_bytes(b"")
288 cue_item = _make_cue_item(tmp_path, "", name="different.cue")
289 provider = _make_provider(base_path=str(tmp_path))
290 cue_sheet = MagicMock(file_path="missing.flac")
291 result = await provider._cue.find_audio_file(cue_item, cue_sheet)
292 assert result is None
293
294
295class TestParseCueTracks:
296 """Tests for _parse_cue_tracks end-to-end (with mocked parse_tags/parse_album)."""
297
298 @staticmethod
299 def _wire_provider_for_parse(
300 provider: LocalFileSystemProvider, album: Album | None = None
301 ) -> None:
302 """Install common mocks for _parse_cue_tracks."""
303 provider._parse_album = AsyncMock(return_value=album) # type: ignore[method-assign]
304 provider._parse_artist = AsyncMock( # type: ignore[method-assign]
305 side_effect=lambda name, **_k: Artist(
306 item_id=name,
307 provider=provider.instance_id,
308 name=name,
309 provider_mappings=set(),
310 )
311 )
312
313 @pytest.mark.asyncio
314 async def test_raises_when_no_tracks(self, tmp_path: Path) -> None:
315 """Raises when CUE has no TRACK entries."""
316 cue_item = _make_cue_item(tmp_path, 'TITLE "Empty"\n')
317 provider = _make_provider(base_path=str(tmp_path))
318 provider._parse_album = AsyncMock(return_value=None) # type: ignore[method-assign]
319 with pytest.raises(InvalidDataError):
320 await provider._cue.parse_tracks(cue_item)
321
322 @pytest.mark.asyncio
323 async def test_raises_when_audio_missing(self, tmp_path: Path) -> None:
324 """Raises when the CUE-referenced audio file cannot be located."""
325 cue_item = _make_cue_item(
326 tmp_path,
327 'FILE "missing.flac" WAVE\n TRACK 01 AUDIO\n TITLE "x"\n INDEX 01 00:00:00\n',
328 )
329 provider = _make_provider(base_path=str(tmp_path))
330 provider._parse_album = AsyncMock(return_value=None) # type: ignore[method-assign]
331 with pytest.raises(MediaNotFoundError):
332 await provider._cue.parse_tracks(cue_item)
333
334 @pytest.mark.asyncio
335 async def test_raises_when_audio_has_no_duration(self, tmp_path: Path) -> None:
336 """Raises when the referenced audio file has no usable duration."""
337 audio_file = tmp_path / "album.flac"
338 audio_file.write_bytes(b"")
339 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
340 provider = _make_provider(base_path=str(tmp_path))
341 provider._parse_album = AsyncMock(return_value=None) # type: ignore[method-assign]
342 with (
343 patch(
344 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
345 AsyncMock(return_value=_make_audio_tags(duration=0.0)),
346 ),
347 pytest.raises(InvalidDataError),
348 ):
349 await provider._cue.parse_tracks(cue_item)
350
351 @pytest.mark.asyncio
352 async def test_builds_tracks_with_cue_metadata(self, tmp_path: Path) -> None:
353 """Builds tracks with cue metadata."""
354 # CUE with 3 tracks, audio file exists, audio tags have a different album/artist
355 audio_file = tmp_path / "album.flac"
356 audio_file.write_bytes(b"")
357 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
358 provider = _make_provider(base_path=str(tmp_path))
359 # audio has different album+albumartist+year; CUE should override
360 tags = _make_audio_tags(
361 duration=900.0,
362 album="Different Album From Tag",
363 albumartist="Different Artist From Tag",
364 )
365 album = Album(
366 item_id="a1",
367 provider=provider.instance_id,
368 name="Live at the BBC",
369 provider_mappings=set(),
370 )
371 self._wire_provider_for_parse(provider, album)
372
373 with patch(
374 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
375 AsyncMock(return_value=tags),
376 ):
377 tracks = await provider._cue.parse_tracks(cue_item)
378
379 assert len(tracks) == 3
380 # CUE TITLE overrode audio tag for album name
381 assert tags.tags["album"] == "Live at the BBC"
382 # CUE top-level PERFORMER overrode audio albumartist, stored as the plural
383 # multi-value form (list at runtime, though tags is typed str-valued)
384 albumartists_value: object = tags.tags["albumartists"]
385 assert albumartists_value == ["Dire Straits"]
386 assert "albumartist" not in tags.tags
387 # per-track names from CUE
388 assert tracks[0].name == "Down to the Waterline"
389 assert tracks[1].name == "Six Blade Knife"
390 assert tracks[2].name == "Water of Love"
391 # track numbers preserved
392 assert [t.track_number for t in tracks] == [1, 2, 3]
393 # item_ids are synthetic and distinct
394 ids = [t.item_id for t in tracks]
395 assert len(set(ids)) == 3
396 for track, num in zip(tracks, [1, 2, 3], strict=True):
397 assert track.item_id == make_cue_track_id(cue_item.relative_path, num)
398 mapping = next(iter(track.provider_mappings))
399 assert mapping.details == cue_metadata_checksum(cue_item.checksum)
400 # ISRC from CUE propagates to each track
401 for track in tracks:
402 isrcs = [v for k, v in track.external_ids if k == ExternalID.ISRC]
403 assert len(isrcs) == 1
404 assert isrcs[0].startswith("GBAMU78")
405
406 @pytest.mark.asyncio
407 async def test_track_performer_registers_cue_sheet_as_representative(
408 self, tmp_path: Path
409 ) -> None:
410 """
411 A per-track PERFORMER (and the album artist) register the CUE sheet, not the audio.
412
413 The companion audio file is absorbed into CUE tracks and is never itself a synced
414 item, so only the CUE sheet's own path can later be re-queued to reparse a changed
415 artist.nfo/image for these performers.
416 """
417 audio_file = tmp_path / "album.flac"
418 audio_file.write_bytes(b"")
419 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
420 provider = _make_provider(base_path=str(tmp_path))
421 tags = _make_audio_tags(duration=900.0)
422 album = Album(
423 item_id="a1",
424 provider=provider.instance_id,
425 name="Live at the BBC",
426 provider_mappings=set(),
427 )
428 self._wire_provider_for_parse(provider, album)
429
430 with patch(
431 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
432 AsyncMock(return_value=tags),
433 ):
434 await provider._cue.parse_tracks(cue_item)
435
436 # every _parse_artist call made while building the per-track performers must carry
437 # the CUE sheet's own path, never the companion audio file's
438 parse_artist_mock = cast("AsyncMock", provider._parse_artist)
439 for call in parse_artist_mock.await_args_list:
440 assert call.kwargs.get("representative_track") == cue_item.relative_path
441
442 @pytest.mark.asyncio
443 async def test_track_durations(self, tmp_path: Path) -> None:
444 """Track durations."""
445 audio_file = tmp_path / "album.flac"
446 audio_file.write_bytes(b"")
447 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
448 provider = _make_provider(base_path=str(tmp_path))
449 # total_duration = 900s (15min)
450 tags = _make_audio_tags(duration=900.0, album="Live at the BBC")
451 self._wire_provider_for_parse(provider)
452
453 with patch(
454 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
455 AsyncMock(return_value=tags),
456 ):
457 tracks = await provider._cue.parse_tracks(cue_item)
458
459 # Track 1: 00:00:00 to 04:10:40 = ~250.53s
460 # Track 2: 04:10:40 to 07:58:05 = ~227.4s
461 # Track 3: 07:58:05 to 900 = ~421.93s
462 assert tracks[0].duration == round(4 * 60 + 10 + 40 / 75)
463 assert tracks[1].duration == round((7 * 60 + 58 + 5 / 75) - (4 * 60 + 10 + 40 / 75))
464 assert tracks[2].duration == round(900.0 - (7 * 60 + 58 + 5 / 75))
465
466 @pytest.mark.asyncio
467 async def test_honors_disc_number_tag(self, tmp_path: Path) -> None:
468 """Honors disc number tag."""
469 audio_file = tmp_path / "album.flac"
470 audio_file.write_bytes(b"")
471 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
472 provider = _make_provider(base_path=str(tmp_path))
473 tags = _make_audio_tags(duration=900.0, album="Live at the BBC", disc="2")
474 self._wire_provider_for_parse(provider)
475
476 with patch(
477 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
478 AsyncMock(return_value=tags),
479 ):
480 tracks = await provider._cue.parse_tracks(cue_item)
481
482 assert all(t.disc_number == 2 for t in tracks)
483
484 @pytest.mark.asyncio
485 async def test_skips_track_missing_title(self, tmp_path: Path) -> None:
486 """Skips track missing title."""
487 audio_file = tmp_path / "album.flac"
488 audio_file.write_bytes(b"")
489 cue_text = (
490 'PERFORMER "X"\n'
491 'TITLE "Album"\n'
492 'FILE "album.flac" WAVE\n'
493 " TRACK 01 AUDIO\n"
494 ' TITLE "Real Track"\n'
495 " INDEX 01 00:00:00\n"
496 " TRACK 02 AUDIO\n"
497 " INDEX 01 02:00:00\n" # no TITLE
498 )
499 cue_item = _make_cue_item(tmp_path, cue_text)
500 provider = _make_provider(base_path=str(tmp_path))
501 tags = _make_audio_tags(duration=300.0, album="Album")
502 self._wire_provider_for_parse(provider)
503
504 with patch(
505 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
506 AsyncMock(return_value=tags),
507 ):
508 tracks = await provider._cue.parse_tracks(cue_item)
509
510 assert len(tracks) == 1
511 assert tracks[0].name == "Real Track"
512 # a warning should have been emitted for the skipped track
513 warning_msgs = [str(c) for c in provider.logger.warning.call_args_list] # type: ignore[attr-defined]
514 assert any("TITLE" in msg for msg in warning_msgs)
515
516 @pytest.mark.asyncio
517 async def test_track_artist_falls_back_to_album_performer(self, tmp_path: Path) -> None:
518 """Track artist falls back to album performer."""
519 audio_file = tmp_path / "album.flac"
520 audio_file.write_bytes(b"")
521 # no per-track PERFORMER, only top-level
522 cue_text = (
523 'PERFORMER "Band"\n'
524 'TITLE "Album"\n'
525 'FILE "album.flac" WAVE\n'
526 " TRACK 01 AUDIO\n"
527 ' TITLE "T1"\n'
528 " INDEX 01 00:00:00\n"
529 )
530 cue_item = _make_cue_item(tmp_path, cue_text)
531 provider = _make_provider(base_path=str(tmp_path))
532 tags = _make_audio_tags(duration=300.0, album="Album")
533 self._wire_provider_for_parse(provider)
534
535 with patch(
536 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
537 AsyncMock(return_value=tags),
538 ):
539 tracks = await provider._cue.parse_tracks(cue_item)
540
541 assert len(tracks) == 1
542 assert [a.name for a in tracks[0].artists] == ["Band"]
543
544 @pytest.mark.asyncio
545 async def test_track_artist_falls_back_to_unknown_when_no_performer(
546 self, tmp_path: Path
547 ) -> None:
548 """No PERFORMER at sheet or track level falls back to the [unknown] artist."""
549 audio_file = tmp_path / "album.flac"
550 audio_file.write_bytes(b"")
551 cue_text = (
552 'TITLE "Album"\n'
553 'FILE "album.flac" WAVE\n'
554 " TRACK 01 AUDIO\n"
555 ' TITLE "T1"\n'
556 " INDEX 01 00:00:00\n"
557 )
558 cue_item = _make_cue_item(tmp_path, cue_text)
559 provider = _make_provider(base_path=str(tmp_path))
560 tags = _make_audio_tags(duration=300.0, album="Album")
561 self._wire_provider_for_parse(provider)
562
563 with patch(
564 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
565 AsyncMock(return_value=tags),
566 ):
567 tracks = await provider._cue.parse_tracks(cue_item)
568
569 assert len(tracks) == 1
570 assert [a.name for a in tracks[0].artists] == [UNKNOWN_ARTIST]
571
572 @pytest.mark.asyncio
573 async def test_multi_line_performer_yields_multiple_artists(self, tmp_path: Path) -> None:
574 """Repeated PERFORMER lines produce one Artist each (Vorbis multi-field style)."""
575 audio_file = tmp_path / "album.flac"
576 audio_file.write_bytes(b"")
577 cue_text = (
578 'TITLE "Split"\n'
579 'FILE "album.flac" WAVE\n'
580 " TRACK 01 AUDIO\n"
581 ' TITLE "T1"\n'
582 ' PERFORMER "AC/DC"\n'
583 ' PERFORMER "Queen"\n'
584 " INDEX 01 00:00:00\n"
585 )
586 cue_item = _make_cue_item(tmp_path, cue_text)
587 provider = _make_provider(base_path=str(tmp_path))
588 tags = _make_audio_tags(duration=300.0, album="Split")
589 self._wire_provider_for_parse(provider)
590
591 with patch(
592 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
593 AsyncMock(return_value=tags),
594 ):
595 tracks = await provider._cue.parse_tracks(cue_item)
596
597 assert len(tracks) == 1
598 # "AC/DC" is preserved intact, not split on the slash
599 assert [a.name for a in tracks[0].artists] == ["AC/DC", "Queen"]
600
601 @pytest.mark.asyncio
602 async def test_recording_and_releasetrack_mbids_mapped_distinctly(self, tmp_path: Path) -> None:
603 """REM MUSICBRAINZ_RECORDINGID â MB_RECORDING / .mbid; REM MUSICBRAINZ_TRACKID â MB_TRACK."""
604 audio_file = tmp_path / "album.flac"
605 audio_file.write_bytes(b"")
606 recording_mbid = "11111111-1111-1111-1111-111111111111"
607 releasetrack_mbid = "22222222-2222-2222-2222-222222222222"
608 cue_text = (
609 'TITLE "Album"\n'
610 'FILE "album.flac" WAVE\n'
611 " TRACK 01 AUDIO\n"
612 ' TITLE "T1"\n'
613 f" REM MUSICBRAINZ_RECORDINGID {recording_mbid}\n"
614 f" REM MUSICBRAINZ_TRACKID {releasetrack_mbid}\n"
615 " INDEX 01 00:00:00\n"
616 )
617 cue_item = _make_cue_item(tmp_path, cue_text)
618 provider = _make_provider(base_path=str(tmp_path))
619 tags = _make_audio_tags(duration=300.0, album="Album")
620 self._wire_provider_for_parse(provider)
621
622 with patch(
623 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
624 AsyncMock(return_value=tags),
625 ):
626 tracks = await provider._cue.parse_tracks(cue_item)
627
628 assert len(tracks) == 1
629 assert (ExternalID.MB_RECORDING, recording_mbid) in tracks[0].external_ids
630 assert (ExternalID.MB_TRACK, releasetrack_mbid) in tracks[0].external_ids
631 assert tracks[0].mbid == recording_mbid
632
633 @pytest.mark.asyncio
634 async def test_aligned_track_artist_metadata(self, tmp_path: Path) -> None:
635 """REM ARTISTSORT / REM MUSICBRAINZ_ARTISTID align by index with PERFORMER."""
636 audio_file = tmp_path / "album.flac"
637 audio_file.write_bytes(b"")
638 cue_text = (
639 'TITLE "Album"\n'
640 'FILE "album.flac" WAVE\n'
641 " TRACK 01 AUDIO\n"
642 ' TITLE "T1"\n'
643 ' PERFORMER "First Artist"\n'
644 ' PERFORMER "Second Artist"\n'
645 ' REM ARTISTSORT "Artist, First"\n'
646 ' REM ARTISTSORT "Artist, Second"\n'
647 " REM MUSICBRAINZ_ARTISTID 11111111-1111-1111-1111-111111111111\n"
648 " REM MUSICBRAINZ_ARTISTID 22222222-2222-2222-2222-222222222222\n"
649 " INDEX 01 00:00:00\n"
650 )
651 cue_item = _make_cue_item(tmp_path, cue_text)
652 provider = _make_provider(base_path=str(tmp_path))
653 tags = _make_audio_tags(duration=300.0, album="Album")
654 self._wire_provider_for_parse(provider)
655 # override _parse_artist to capture the sort_name/mbid args passed per artist
656 captured: list[dict[str, str | None]] = []
657
658 async def _capture(
659 name: str, sort_name: str | None = None, mbid: str | None = None, **_k: object
660 ) -> Artist:
661 captured.append({"name": name, "sort_name": sort_name, "mbid": mbid})
662 return Artist(
663 item_id=name,
664 provider=provider.instance_id,
665 name=name,
666 sort_name=sort_name,
667 provider_mappings=set(),
668 )
669
670 provider._parse_artist = AsyncMock(side_effect=_capture) # type: ignore[method-assign]
671
672 with patch(
673 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
674 AsyncMock(return_value=tags),
675 ):
676 tracks = await provider._cue.parse_tracks(cue_item)
677
678 assert len(tracks) == 1
679 assert captured == [
680 {
681 "name": "First Artist",
682 "sort_name": "Artist, First",
683 "mbid": "11111111-1111-1111-1111-111111111111",
684 },
685 {
686 "name": "Second Artist",
687 "sort_name": "Artist, Second",
688 "mbid": "22222222-2222-2222-2222-222222222222",
689 },
690 ]
691
692 @pytest.mark.asyncio
693 async def test_track_level_descriptive_fields(self, tmp_path: Path) -> None:
694 """REM COPYRIGHT / GROUPING / COMMENT / ITUNESADVISORY / TITLESORT populate track metadata."""
695 audio_file = tmp_path / "album.flac"
696 audio_file.write_bytes(b"")
697 cue_text = (
698 'TITLE "Album"\n'
699 'FILE "album.flac" WAVE\n'
700 " TRACK 01 AUDIO\n"
701 ' TITLE "Song, The"\n'
702 ' REM TITLESORT "Song, The"\n'
703 ' REM COPYRIGHT "(c) 2024 Label"\n'
704 ' REM GROUPING "Movement I"\n'
705 ' REM COMMENT "Live at Wembley"\n'
706 " REM ITUNESADVISORY 1\n"
707 " INDEX 01 00:00:00\n"
708 )
709 cue_item = _make_cue_item(tmp_path, cue_text)
710 provider = _make_provider(base_path=str(tmp_path))
711 tags = _make_audio_tags(duration=300.0, album="Album")
712 self._wire_provider_for_parse(provider)
713
714 with patch(
715 "music_assistant.providers.filesystem_local.cue.async_parse_tags",
716 AsyncMock(return_value=tags),
717 ):
718 tracks = await provider._cue.parse_tracks(cue_item)
719
720 assert len(tracks) == 1
721 track = tracks[0]
722 assert track.sort_name == "Song, The"
723 assert track.metadata.copyright == "(c) 2024 Label"
724 assert track.metadata.grouping == "Movement I"
725 assert track.metadata.description == "Live at Wembley"
726 assert track.metadata.explicit is True
727
728
729class TestGetStreamDetailsForCueTrack:
730 """Tests for _get_stream_details_for_cue_track."""
731
732 @pytest.mark.asyncio
733 async def test_invalid_id_raises(self) -> None:
734 """Invalid id raises."""
735 provider = _make_provider()
736 with pytest.raises(InvalidDataError):
737 await provider._cue.get_stream_details("not_a_cue_id.flac")
738
739 @pytest.mark.asyncio
740 async def test_not_in_library_raises(self, tmp_path: Path) -> None:
741 """Track not in library raises."""
742 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
743 provider = _make_provider(base_path=str(tmp_path))
744 provider.resolve = AsyncMock(return_value=cue_item) # type: ignore[method-assign]
745 provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None) # type: ignore[method-assign]
746 item_id = make_cue_track_id(cue_item.relative_path, 1)
747 with pytest.raises(MediaNotFoundError):
748 await provider._cue.get_stream_details(item_id)
749
750 @pytest.mark.asyncio
751 async def test_missing_audio_raises(self, tmp_path: Path) -> None:
752 """Missing audio raises."""
753 cue_item = _make_cue_item(
754 tmp_path,
755 'FILE "missing.flac" WAVE\n TRACK 01 AUDIO\n TITLE "x"\n INDEX 01 00:00:00\n',
756 )
757 provider = _make_provider(base_path=str(tmp_path))
758 provider.resolve = AsyncMock(return_value=cue_item) # type: ignore[method-assign]
759 item_id = make_cue_track_id(cue_item.relative_path, 1)
760 _stub_library_track(provider, item_id)
761 with pytest.raises(MediaNotFoundError):
762 await provider._cue.get_stream_details(item_id)
763
764 @pytest.mark.asyncio
765 async def test_unknown_track_number_raises(self, tmp_path: Path) -> None:
766 """Unknown track number raises."""
767 audio_file = tmp_path / "album.flac"
768 audio_file.write_bytes(b"")
769 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
770 provider = _make_provider(base_path=str(tmp_path))
771 provider.resolve = AsyncMock(return_value=cue_item) # type: ignore[method-assign]
772 # request track 99 which isn't in the CUE
773 item_id = make_cue_track_id(cue_item.relative_path, 99)
774 _stub_library_track(provider, item_id)
775 with pytest.raises(MediaNotFoundError):
776 await provider._cue.get_stream_details(item_id)
777
778 @pytest.mark.asyncio
779 async def test_builds_streamdetails_with_offset_and_duration(self, tmp_path: Path) -> None:
780 """Builds streamdetails with offset and duration."""
781 audio_file = tmp_path / "album.flac"
782 audio_file.write_bytes(b"")
783 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
784 provider = _make_provider(base_path=str(tmp_path))
785 provider.resolve = AsyncMock(return_value=cue_item) # type: ignore[method-assign]
786 item_id = make_cue_track_id(cue_item.relative_path, 2)
787 _stub_library_track(provider, item_id, duration=228)
788
789 details = await provider._cue.get_stream_details(item_id)
790
791 assert isinstance(details, StreamDetails)
792 assert details.item_id == item_id
793 assert details.can_seek is True
794 assert details.allow_seek is True
795 assert details.audio_format.content_type == ContentType.PCM_F32LE
796 assert details.duration == 228
797 assert details.data is not None
798 assert details.data["audio_relative_path"] == "album.flac"
799 # Track 2 starts at 04:10:40 = 250.533...
800 expected_start = 4 * 60 + 10 + 40 / 75
801 assert abs(details.data["start_seconds"] - expected_start) < 0.001
802
803
804class TestProcessDeletionsCueBranch:
805 """Tests for _process_deletions routing of CUE-derived track ids."""
806
807 @pytest.mark.asyncio
808 async def test_cue_track_id_routed_to_track_controller(self) -> None:
809 """Cue track id routed to track controller."""
810 provider = _make_provider()
811 controller = MagicMock()
812 controller.get_library_item_by_prov_id = AsyncMock(return_value=None)
813 provider.mass.music.get_controller = MagicMock(return_value=controller) # type: ignore[method-assign]
814
815 cue_track_id = make_cue_track_id("artist/album.cue", 3)
816 await provider._process_deletions({cue_track_id})
817
818 # must have consulted a controller (the track controller)
819 assert provider.mass.music.get_controller.called
820
821
822class TestGetTrackCueBranch:
823 """Test get_track's CUE-id branch."""
824
825 @pytest.mark.asyncio
826 async def test_returns_matching_cue_track(self, tmp_path: Path) -> None:
827 """Returns matching cue track."""
828 audio_file = tmp_path / "album.flac"
829 audio_file.write_bytes(b"")
830 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
831 provider = _make_provider(base_path=str(tmp_path))
832 provider.resolve = AsyncMock(return_value=cue_item) # type: ignore[method-assign]
833
834 # mock _parse_cue_tracks to return three synthetic tracks
835 def fake_track(num: int) -> Track:
836 return Track(
837 item_id=make_cue_track_id(cue_item.relative_path, num),
838 provider=provider.instance_id,
839 name=f"Track {num}",
840 provider_mappings=set(),
841 )
842
843 provider._cue.parse_tracks = AsyncMock( # type: ignore[method-assign]
844 return_value=[fake_track(1), fake_track(2), fake_track(3)]
845 )
846 item_id = make_cue_track_id(cue_item.relative_path, 2)
847 track = await provider.get_track(item_id)
848 assert track.item_id == item_id
849 assert track.name == "Track 2"
850
851 @pytest.mark.asyncio
852 async def test_missing_cue_track_raises(self, tmp_path: Path) -> None:
853 """Missing cue track raises."""
854 cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
855 provider = _make_provider(base_path=str(tmp_path))
856 provider.resolve = AsyncMock(return_value=cue_item) # type: ignore[method-assign]
857 provider._cue.parse_tracks = AsyncMock(return_value=[]) # type: ignore[method-assign]
858 item_id = make_cue_track_id(cue_item.relative_path, 1)
859 with pytest.raises(MediaNotFoundError):
860 await provider.get_track(item_id)
861
862
863class TestClassifyScanItemCue:
864 """
865 Sync-walker classification for CUE files.
866
867 Guards the edit-resync path: a CUE's previous checksum lives under synthetic
868 per-track ids in provider_mappings, never under the CUE path itself. The
869 scan reverse-derives a path-keyed map so an unchanged CUE is recognised and
870 an edited CUE forwards its prior checksum, which in turn makes the library
871 write use overwrite_existing=True.
872 """
873
874 @staticmethod
875 def _cue_item(checksum: str) -> FileSystemItem:
876 return FileSystemItem(
877 filename="album.cue",
878 relative_path="album.cue",
879 absolute_path="/music/album.cue",
880 is_dir=False,
881 checksum=checksum,
882 file_size=100,
883 )
884
885 @staticmethod
886 def _classify(
887 provider: LocalFileSystemProvider,
888 item: FileSystemItem,
889 *,
890 cue_file_checksums: dict[str, set[str]] | None = None,
891 ) -> tuple[
892 list[tuple[FileSystemItem, str | None]],
893 list[FileSystemItem],
894 set[str],
895 set[str],
896 ]:
897 items_to_process: list[tuple[FileSystemItem, str | None]] = []
898 unchanged_cue_items: list[FileSystemItem] = []
899 cur_filenames: set[str] = set()
900 cue_stems: set[str] = set()
901 provider._classify_scan_item(
902 item,
903 file_checksums={},
904 cue_file_checksums=cue_file_checksums or {},
905 cur_filenames=cur_filenames,
906 items_to_process=items_to_process,
907 unchanged_cue_items=unchanged_cue_items,
908 cue_stems=cue_stems,
909 ignore_album_playlists=False,
910 metadata_files=[],
911 )
912 return items_to_process, unchanged_cue_items, cur_filenames, cue_stems
913
914 def test_unchanged_cue_routes_to_unchanged_bucket(self) -> None:
915 """Matching checksum: CUE is marked present and not re-processed."""
916 provider = _make_provider()
917 cue_item = self._cue_item("checksum-v1")
918 items, unchanged, cur, stems = self._classify(
919 provider,
920 cue_item,
921 cue_file_checksums={"album.cue": {cue_metadata_checksum("checksum-v1")}},
922 )
923 assert items == []
924 assert unchanged == [cue_item]
925 assert cur == {"album.cue"}
926 assert stems == {"/music/album"}
927
928 def test_edited_cue_forwards_prior_checksum(self) -> None:
929 """Changed checksum: prior value is forwarded so downstream overwrite=True."""
930 provider = _make_provider()
931 cue_item = self._cue_item("checksum-v2")
932 items, unchanged, _, _ = self._classify(
933 provider,
934 cue_item,
935 cue_file_checksums={"album.cue": {cue_metadata_checksum("checksum-v1")}},
936 )
937 assert items == [(cue_item, cue_metadata_checksum("checksum-v1"))]
938 assert unchanged == []
939
940 def test_legacy_checksum_forces_metadata_refresh(self) -> None:
941 """An unversioned mapping is reprocessed even when the file is unchanged."""
942 provider = _make_provider()
943 cue_item = self._cue_item("checksum-v1")
944 items, unchanged, _, _ = self._classify(
945 provider,
946 cue_item,
947 cue_file_checksums={"album.cue": {"checksum-v1"}},
948 )
949 assert items == [(cue_item, "checksum-v1")]
950 assert unchanged == []
951
952 def test_mixed_checksums_force_metadata_refresh(self) -> None:
953 """A partially refreshed CUE is reprocessed until every track is current."""
954 provider = _make_provider()
955 cue_item = self._cue_item("checksum-v1")
956 previous_checksums = {
957 "checksum-v1",
958 cue_metadata_checksum("checksum-v1"),
959 }
960 items, unchanged, _, _ = self._classify(
961 provider,
962 cue_item,
963 cue_file_checksums={"album.cue": previous_checksums},
964 )
965 assert items == [(cue_item, min(previous_checksums))]
966 assert unchanged == []
967
968 def test_new_cue_has_no_prior_checksum(self) -> None:
969 """First-time ingest: prev_checksum is None, item queued for processing."""
970 provider = _make_provider()
971 cue_item = self._cue_item("checksum-v1")
972 items, unchanged, _, _ = self._classify(
973 provider,
974 cue_item,
975 cue_file_checksums={},
976 )
977 assert items == [(cue_item, None)]
978 assert unchanged == []
979