music-assistant-server

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