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_non_utf8_tolerantly(self, tmp_path: Path) -> None:
195        """Non-UTF-8 bytes are decoded without raising."""
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        # ASCII surroundings are preserved; the non-UTF-8 byte may be replaced
209        # or decoded depending on what chardet detects
210        assert "TITLE" in content
211        assert "ller" in content
212
213
214class TestFindCueAudioFile:
215    """Tests for audio file resolution from a CUE sheet."""
216
217    @pytest.mark.asyncio
218    async def test_matches_file_command(self, tmp_path: Path) -> None:
219        """Matches file command."""
220        (tmp_path / "album.flac").write_bytes(b"")
221        (tmp_path / "other.flac").write_bytes(b"")
222        cue_item = _make_cue_item(tmp_path, 'FILE "album.flac" WAVE\n')
223        provider = _make_provider(base_path=str(tmp_path))
224        cue_sheet = MagicMock(file_path="album.flac")
225        result = await provider._cue.find_audio_file(cue_item, cue_sheet)
226        assert result == "album.flac"
227
228    @pytest.mark.asyncio
229    async def test_same_stem_fallback(self, tmp_path: Path) -> None:
230        """Same stem fallback."""
231        (tmp_path / "album.flac").write_bytes(b"")
232        cue_item = _make_cue_item(tmp_path, "")
233        provider = _make_provider(base_path=str(tmp_path))
234        cue_sheet = MagicMock(file_path=None)
235        result = await provider._cue.find_audio_file(cue_item, cue_sheet)
236        assert result == "album.flac"
237
238    @pytest.mark.asyncio
239    async def test_returns_none_when_file_missing_and_stem_mismatch(self, tmp_path: Path) -> None:
240        """Returns None when neither FILE nor same-stem match locates the audio file."""
241        (tmp_path / "onlyone.flac").write_bytes(b"")
242        cue_item = _make_cue_item(tmp_path, "", name="different.cue")
243        provider = _make_provider(base_path=str(tmp_path))
244        cue_sheet = MagicMock(file_path="missing.flac")
245        result = await provider._cue.find_audio_file(cue_item, cue_sheet)
246        assert result is None
247
248
249class TestParseCueTracks:
250    """Tests for _parse_cue_tracks end-to-end (with mocked parse_tags/parse_album)."""
251
252    @staticmethod
253    def _wire_provider_for_parse(
254        provider: LocalFileSystemProvider, album: Album | None = None
255    ) -> None:
256        """Install common mocks for _parse_cue_tracks."""
257        provider._parse_album = AsyncMock(return_value=album)  # type: ignore[method-assign]
258        provider._parse_artist = AsyncMock(  # type: ignore[method-assign]
259            side_effect=lambda name, **_k: Artist(
260                item_id=name,
261                provider=provider.instance_id,
262                name=name,
263                provider_mappings=set(),
264            )
265        )
266
267    @pytest.mark.asyncio
268    async def test_raises_when_no_tracks(self, tmp_path: Path) -> None:
269        """Raises when CUE has no TRACK entries."""
270        cue_item = _make_cue_item(tmp_path, 'TITLE "Empty"\n')
271        provider = _make_provider(base_path=str(tmp_path))
272        provider._parse_album = AsyncMock(return_value=None)  # type: ignore[method-assign]
273        with pytest.raises(InvalidDataError):
274            await provider._cue.parse_tracks(cue_item)
275
276    @pytest.mark.asyncio
277    async def test_raises_when_audio_missing(self, tmp_path: Path) -> None:
278        """Raises when the CUE-referenced audio file cannot be located."""
279        cue_item = _make_cue_item(
280            tmp_path,
281            'FILE "missing.flac" WAVE\n  TRACK 01 AUDIO\n    TITLE "x"\n    INDEX 01 00:00:00\n',
282        )
283        provider = _make_provider(base_path=str(tmp_path))
284        provider._parse_album = AsyncMock(return_value=None)  # type: ignore[method-assign]
285        with pytest.raises(MediaNotFoundError):
286            await provider._cue.parse_tracks(cue_item)
287
288    @pytest.mark.asyncio
289    async def test_raises_when_audio_has_no_duration(self, tmp_path: Path) -> None:
290        """Raises when the referenced audio file has no usable duration."""
291        audio_file = tmp_path / "album.flac"
292        audio_file.write_bytes(b"")
293        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
294        provider = _make_provider(base_path=str(tmp_path))
295        provider._parse_album = AsyncMock(return_value=None)  # type: ignore[method-assign]
296        with (
297            patch(
298                "music_assistant.providers.filesystem_local.cue.async_parse_tags",
299                AsyncMock(return_value=_make_audio_tags(duration=0.0)),
300            ),
301            pytest.raises(InvalidDataError),
302        ):
303            await provider._cue.parse_tracks(cue_item)
304
305    @pytest.mark.asyncio
306    async def test_builds_tracks_with_cue_metadata(self, tmp_path: Path) -> None:
307        """Builds tracks with cue metadata."""
308        # CUE with 3 tracks, audio file exists, audio tags have a different album/artist
309        audio_file = tmp_path / "album.flac"
310        audio_file.write_bytes(b"")
311        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
312        provider = _make_provider(base_path=str(tmp_path))
313        # audio has different album+albumartist+year; CUE should override
314        tags = _make_audio_tags(
315            duration=900.0,
316            album="Different Album From Tag",
317            albumartist="Different Artist From Tag",
318        )
319        album = Album(
320            item_id="a1",
321            provider=provider.instance_id,
322            name="Live at the BBC",
323            provider_mappings=set(),
324        )
325        self._wire_provider_for_parse(provider, album)
326
327        with patch(
328            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
329            AsyncMock(return_value=tags),
330        ):
331            tracks = await provider._cue.parse_tracks(cue_item)
332
333        assert len(tracks) == 3
334        # CUE TITLE overrode audio tag for album name
335        assert tags.tags["album"] == "Live at the BBC"
336        # CUE top-level PERFORMER overrode audio albumartist, stored as the plural
337        # multi-value form (list at runtime, though tags is typed str-valued)
338        albumartists_value: object = tags.tags["albumartists"]
339        assert albumartists_value == ["Dire Straits"]
340        assert "albumartist" not in tags.tags
341        # per-track names from CUE
342        assert tracks[0].name == "Down to the Waterline"
343        assert tracks[1].name == "Six Blade Knife"
344        assert tracks[2].name == "Water of Love"
345        # track numbers preserved
346        assert [t.track_number for t in tracks] == [1, 2, 3]
347        # item_ids are synthetic and distinct
348        ids = [t.item_id for t in tracks]
349        assert len(set(ids)) == 3
350        for track, num in zip(tracks, [1, 2, 3], strict=True):
351            assert track.item_id == make_cue_track_id(cue_item.relative_path, num)
352        # ISRC from CUE propagates to each track
353        for track in tracks:
354            isrcs = [v for k, v in track.external_ids if k == ExternalID.ISRC]
355            assert len(isrcs) == 1
356            assert isrcs[0].startswith("GBAMU78")
357
358    @pytest.mark.asyncio
359    async def test_track_durations(self, tmp_path: Path) -> None:
360        """Track durations."""
361        audio_file = tmp_path / "album.flac"
362        audio_file.write_bytes(b"")
363        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
364        provider = _make_provider(base_path=str(tmp_path))
365        # total_duration = 900s (15min)
366        tags = _make_audio_tags(duration=900.0, album="Live at the BBC")
367        self._wire_provider_for_parse(provider)
368
369        with patch(
370            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
371            AsyncMock(return_value=tags),
372        ):
373            tracks = await provider._cue.parse_tracks(cue_item)
374
375        # Track 1: 00:00:00 to 04:10:40 = ~250.53s
376        # Track 2: 04:10:40 to 07:58:05 = ~227.4s
377        # Track 3: 07:58:05 to 900 = ~421.93s
378        assert tracks[0].duration == round(4 * 60 + 10 + 40 / 75)
379        assert tracks[1].duration == round((7 * 60 + 58 + 5 / 75) - (4 * 60 + 10 + 40 / 75))
380        assert tracks[2].duration == round(900.0 - (7 * 60 + 58 + 5 / 75))
381
382    @pytest.mark.asyncio
383    async def test_honors_disc_number_tag(self, tmp_path: Path) -> None:
384        """Honors disc number tag."""
385        audio_file = tmp_path / "album.flac"
386        audio_file.write_bytes(b"")
387        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
388        provider = _make_provider(base_path=str(tmp_path))
389        tags = _make_audio_tags(duration=900.0, album="Live at the BBC", disc="2")
390        self._wire_provider_for_parse(provider)
391
392        with patch(
393            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
394            AsyncMock(return_value=tags),
395        ):
396            tracks = await provider._cue.parse_tracks(cue_item)
397
398        assert all(t.disc_number == 2 for t in tracks)
399
400    @pytest.mark.asyncio
401    async def test_skips_track_missing_title(self, tmp_path: Path) -> None:
402        """Skips track missing title."""
403        audio_file = tmp_path / "album.flac"
404        audio_file.write_bytes(b"")
405        cue_text = (
406            'PERFORMER "X"\n'
407            'TITLE "Album"\n'
408            'FILE "album.flac" WAVE\n'
409            "  TRACK 01 AUDIO\n"
410            '    TITLE "Real Track"\n'
411            "    INDEX 01 00:00:00\n"
412            "  TRACK 02 AUDIO\n"
413            "    INDEX 01 02:00:00\n"  # no TITLE
414        )
415        cue_item = _make_cue_item(tmp_path, cue_text)
416        provider = _make_provider(base_path=str(tmp_path))
417        tags = _make_audio_tags(duration=300.0, album="Album")
418        self._wire_provider_for_parse(provider)
419
420        with patch(
421            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
422            AsyncMock(return_value=tags),
423        ):
424            tracks = await provider._cue.parse_tracks(cue_item)
425
426        assert len(tracks) == 1
427        assert tracks[0].name == "Real Track"
428        # a warning should have been emitted for the skipped track
429        warning_msgs = [str(c) for c in provider.logger.warning.call_args_list]  # type: ignore[attr-defined]
430        assert any("TITLE" in msg for msg in warning_msgs)
431
432    @pytest.mark.asyncio
433    async def test_track_artist_falls_back_to_album_performer(self, tmp_path: Path) -> None:
434        """Track artist falls back to album performer."""
435        audio_file = tmp_path / "album.flac"
436        audio_file.write_bytes(b"")
437        # no per-track PERFORMER, only top-level
438        cue_text = (
439            'PERFORMER "Band"\n'
440            'TITLE "Album"\n'
441            'FILE "album.flac" WAVE\n'
442            "  TRACK 01 AUDIO\n"
443            '    TITLE "T1"\n'
444            "    INDEX 01 00:00:00\n"
445        )
446        cue_item = _make_cue_item(tmp_path, cue_text)
447        provider = _make_provider(base_path=str(tmp_path))
448        tags = _make_audio_tags(duration=300.0, album="Album")
449        self._wire_provider_for_parse(provider)
450
451        with patch(
452            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
453            AsyncMock(return_value=tags),
454        ):
455            tracks = await provider._cue.parse_tracks(cue_item)
456
457        assert len(tracks) == 1
458        assert [a.name for a in tracks[0].artists] == ["Band"]
459
460    @pytest.mark.asyncio
461    async def test_track_artist_falls_back_to_unknown_when_no_performer(
462        self, tmp_path: Path
463    ) -> None:
464        """No PERFORMER at sheet or track level falls back to the [unknown] artist."""
465        audio_file = tmp_path / "album.flac"
466        audio_file.write_bytes(b"")
467        cue_text = (
468            'TITLE "Album"\n'
469            'FILE "album.flac" WAVE\n'
470            "  TRACK 01 AUDIO\n"
471            '    TITLE "T1"\n'
472            "    INDEX 01 00:00:00\n"
473        )
474        cue_item = _make_cue_item(tmp_path, cue_text)
475        provider = _make_provider(base_path=str(tmp_path))
476        tags = _make_audio_tags(duration=300.0, album="Album")
477        self._wire_provider_for_parse(provider)
478
479        with patch(
480            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
481            AsyncMock(return_value=tags),
482        ):
483            tracks = await provider._cue.parse_tracks(cue_item)
484
485        assert len(tracks) == 1
486        assert [a.name for a in tracks[0].artists] == [UNKNOWN_ARTIST]
487
488    @pytest.mark.asyncio
489    async def test_multi_line_performer_yields_multiple_artists(self, tmp_path: Path) -> None:
490        """Repeated PERFORMER lines produce one Artist each (Vorbis multi-field style)."""
491        audio_file = tmp_path / "album.flac"
492        audio_file.write_bytes(b"")
493        cue_text = (
494            'TITLE "Split"\n'
495            'FILE "album.flac" WAVE\n'
496            "  TRACK 01 AUDIO\n"
497            '    TITLE "T1"\n'
498            '    PERFORMER "AC/DC"\n'
499            '    PERFORMER "Queen"\n'
500            "    INDEX 01 00:00:00\n"
501        )
502        cue_item = _make_cue_item(tmp_path, cue_text)
503        provider = _make_provider(base_path=str(tmp_path))
504        tags = _make_audio_tags(duration=300.0, album="Split")
505        self._wire_provider_for_parse(provider)
506
507        with patch(
508            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
509            AsyncMock(return_value=tags),
510        ):
511            tracks = await provider._cue.parse_tracks(cue_item)
512
513        assert len(tracks) == 1
514        # "AC/DC" is preserved intact, not split on the slash
515        assert [a.name for a in tracks[0].artists] == ["AC/DC", "Queen"]
516
517    @pytest.mark.asyncio
518    async def test_recording_and_releasetrack_mbids_mapped_distinctly(self, tmp_path: Path) -> None:
519        """REM MUSICBRAINZ_RECORDINGID → MB_RECORDING / .mbid; REM MUSICBRAINZ_TRACKID → MB_TRACK."""
520        audio_file = tmp_path / "album.flac"
521        audio_file.write_bytes(b"")
522        recording_mbid = "11111111-1111-1111-1111-111111111111"
523        releasetrack_mbid = "22222222-2222-2222-2222-222222222222"
524        cue_text = (
525            'TITLE "Album"\n'
526            'FILE "album.flac" WAVE\n'
527            "  TRACK 01 AUDIO\n"
528            '    TITLE "T1"\n'
529            f"    REM MUSICBRAINZ_RECORDINGID {recording_mbid}\n"
530            f"    REM MUSICBRAINZ_TRACKID {releasetrack_mbid}\n"
531            "    INDEX 01 00:00:00\n"
532        )
533        cue_item = _make_cue_item(tmp_path, cue_text)
534        provider = _make_provider(base_path=str(tmp_path))
535        tags = _make_audio_tags(duration=300.0, album="Album")
536        self._wire_provider_for_parse(provider)
537
538        with patch(
539            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
540            AsyncMock(return_value=tags),
541        ):
542            tracks = await provider._cue.parse_tracks(cue_item)
543
544        assert len(tracks) == 1
545        assert (ExternalID.MB_RECORDING, recording_mbid) in tracks[0].external_ids
546        assert (ExternalID.MB_TRACK, releasetrack_mbid) in tracks[0].external_ids
547        assert tracks[0].mbid == recording_mbid
548
549    @pytest.mark.asyncio
550    async def test_aligned_track_artist_metadata(self, tmp_path: Path) -> None:
551        """REM ARTISTSORT / REM MUSICBRAINZ_ARTISTID align by index with PERFORMER."""
552        audio_file = tmp_path / "album.flac"
553        audio_file.write_bytes(b"")
554        cue_text = (
555            'TITLE "Album"\n'
556            'FILE "album.flac" WAVE\n'
557            "  TRACK 01 AUDIO\n"
558            '    TITLE "T1"\n'
559            '    PERFORMER "First Artist"\n'
560            '    PERFORMER "Second Artist"\n'
561            '    REM ARTISTSORT "Artist, First"\n'
562            '    REM ARTISTSORT "Artist, Second"\n'
563            "    REM MUSICBRAINZ_ARTISTID 11111111-1111-1111-1111-111111111111\n"
564            "    REM MUSICBRAINZ_ARTISTID 22222222-2222-2222-2222-222222222222\n"
565            "    INDEX 01 00:00:00\n"
566        )
567        cue_item = _make_cue_item(tmp_path, cue_text)
568        provider = _make_provider(base_path=str(tmp_path))
569        tags = _make_audio_tags(duration=300.0, album="Album")
570        self._wire_provider_for_parse(provider)
571        # override _parse_artist to capture the sort_name/mbid args passed per artist
572        captured: list[dict[str, str | None]] = []
573
574        async def _capture(
575            name: str, sort_name: str | None = None, mbid: str | None = None, **_k: object
576        ) -> Artist:
577            captured.append({"name": name, "sort_name": sort_name, "mbid": mbid})
578            return Artist(
579                item_id=name,
580                provider=provider.instance_id,
581                name=name,
582                sort_name=sort_name,
583                provider_mappings=set(),
584            )
585
586        provider._parse_artist = AsyncMock(side_effect=_capture)  # type: ignore[method-assign]
587
588        with patch(
589            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
590            AsyncMock(return_value=tags),
591        ):
592            tracks = await provider._cue.parse_tracks(cue_item)
593
594        assert len(tracks) == 1
595        assert captured == [
596            {
597                "name": "First Artist",
598                "sort_name": "Artist, First",
599                "mbid": "11111111-1111-1111-1111-111111111111",
600            },
601            {
602                "name": "Second Artist",
603                "sort_name": "Artist, Second",
604                "mbid": "22222222-2222-2222-2222-222222222222",
605            },
606        ]
607
608    @pytest.mark.asyncio
609    async def test_track_level_descriptive_fields(self, tmp_path: Path) -> None:
610        """REM COPYRIGHT / GROUPING / COMMENT / ITUNESADVISORY / TITLESORT populate track metadata."""
611        audio_file = tmp_path / "album.flac"
612        audio_file.write_bytes(b"")
613        cue_text = (
614            'TITLE "Album"\n'
615            'FILE "album.flac" WAVE\n'
616            "  TRACK 01 AUDIO\n"
617            '    TITLE "Song, The"\n'
618            '    REM TITLESORT "Song, The"\n'
619            '    REM COPYRIGHT "(c) 2024 Label"\n'
620            '    REM GROUPING "Movement I"\n'
621            '    REM COMMENT "Live at Wembley"\n'
622            "    REM ITUNESADVISORY 1\n"
623            "    INDEX 01 00:00:00\n"
624        )
625        cue_item = _make_cue_item(tmp_path, cue_text)
626        provider = _make_provider(base_path=str(tmp_path))
627        tags = _make_audio_tags(duration=300.0, album="Album")
628        self._wire_provider_for_parse(provider)
629
630        with patch(
631            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
632            AsyncMock(return_value=tags),
633        ):
634            tracks = await provider._cue.parse_tracks(cue_item)
635
636        assert len(tracks) == 1
637        track = tracks[0]
638        assert track.sort_name == "Song, The"
639        assert track.metadata.copyright == "(c) 2024 Label"
640        assert track.metadata.grouping == "Movement I"
641        assert track.metadata.description == "Live at Wembley"
642        assert track.metadata.explicit is True
643
644
645class TestGetStreamDetailsForCueTrack:
646    """Tests for _get_stream_details_for_cue_track."""
647
648    @pytest.mark.asyncio
649    async def test_invalid_id_raises(self) -> None:
650        """Invalid id raises."""
651        provider = _make_provider()
652        with pytest.raises(InvalidDataError):
653            await provider._cue.get_stream_details("not_a_cue_id.flac")
654
655    @pytest.mark.asyncio
656    async def test_not_in_library_raises(self, tmp_path: Path) -> None:
657        """Track not in library raises."""
658        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
659        provider = _make_provider(base_path=str(tmp_path))
660        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
661        provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)  # type: ignore[method-assign]
662        item_id = make_cue_track_id(cue_item.relative_path, 1)
663        with pytest.raises(MediaNotFoundError):
664            await provider._cue.get_stream_details(item_id)
665
666    @pytest.mark.asyncio
667    async def test_missing_audio_raises(self, tmp_path: Path) -> None:
668        """Missing audio raises."""
669        cue_item = _make_cue_item(
670            tmp_path,
671            'FILE "missing.flac" WAVE\n  TRACK 01 AUDIO\n    TITLE "x"\n    INDEX 01 00:00:00\n',
672        )
673        provider = _make_provider(base_path=str(tmp_path))
674        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
675        item_id = make_cue_track_id(cue_item.relative_path, 1)
676        _stub_library_track(provider, item_id)
677        with pytest.raises(MediaNotFoundError):
678            await provider._cue.get_stream_details(item_id)
679
680    @pytest.mark.asyncio
681    async def test_unknown_track_number_raises(self, tmp_path: Path) -> None:
682        """Unknown track number raises."""
683        audio_file = tmp_path / "album.flac"
684        audio_file.write_bytes(b"")
685        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
686        provider = _make_provider(base_path=str(tmp_path))
687        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
688        # request track 99 which isn't in the CUE
689        item_id = make_cue_track_id(cue_item.relative_path, 99)
690        _stub_library_track(provider, item_id)
691        with pytest.raises(MediaNotFoundError):
692            await provider._cue.get_stream_details(item_id)
693
694    @pytest.mark.asyncio
695    async def test_builds_streamdetails_with_offset_and_duration(self, tmp_path: Path) -> None:
696        """Builds streamdetails with offset and duration."""
697        audio_file = tmp_path / "album.flac"
698        audio_file.write_bytes(b"")
699        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
700        provider = _make_provider(base_path=str(tmp_path))
701        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
702        item_id = make_cue_track_id(cue_item.relative_path, 2)
703        _stub_library_track(provider, item_id, duration=228)
704
705        details = await provider._cue.get_stream_details(item_id)
706
707        assert isinstance(details, StreamDetails)
708        assert details.item_id == item_id
709        assert details.can_seek is True
710        assert details.allow_seek is True
711        assert details.audio_format.content_type == ContentType.PCM_F32LE
712        assert details.duration == 228
713        assert details.data is not None
714        assert details.data["audio_relative_path"] == "album.flac"
715        # Track 2 starts at 04:10:40 = 250.533...
716        expected_start = 4 * 60 + 10 + 40 / 75
717        assert abs(details.data["start_seconds"] - expected_start) < 0.001
718
719
720class TestProcessDeletionsCueBranch:
721    """Tests for _process_deletions routing of CUE-derived track ids."""
722
723    @pytest.mark.asyncio
724    async def test_cue_track_id_routed_to_track_controller(self) -> None:
725        """Cue track id routed to track controller."""
726        provider = _make_provider()
727        controller = MagicMock()
728        controller.get_library_item_by_prov_id = AsyncMock(return_value=None)
729        provider.mass.music.get_controller = MagicMock(return_value=controller)  # type: ignore[method-assign]
730
731        cue_track_id = make_cue_track_id("artist/album.cue", 3)
732        await provider._process_deletions({cue_track_id})
733
734        # must have consulted a controller (the track controller)
735        assert provider.mass.music.get_controller.called
736
737
738class TestGetTrackCueBranch:
739    """Test get_track's CUE-id branch."""
740
741    @pytest.mark.asyncio
742    async def test_returns_matching_cue_track(self, tmp_path: Path) -> None:
743        """Returns matching cue track."""
744        audio_file = tmp_path / "album.flac"
745        audio_file.write_bytes(b"")
746        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
747        provider = _make_provider(base_path=str(tmp_path))
748        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
749
750        # mock _parse_cue_tracks to return three synthetic tracks
751        def fake_track(num: int) -> Track:
752            return Track(
753                item_id=make_cue_track_id(cue_item.relative_path, num),
754                provider=provider.instance_id,
755                name=f"Track {num}",
756                provider_mappings=set(),
757            )
758
759        provider._cue.parse_tracks = AsyncMock(  # type: ignore[method-assign]
760            return_value=[fake_track(1), fake_track(2), fake_track(3)]
761        )
762        item_id = make_cue_track_id(cue_item.relative_path, 2)
763        track = await provider.get_track(item_id)
764        assert track.item_id == item_id
765        assert track.name == "Track 2"
766
767    @pytest.mark.asyncio
768    async def test_missing_cue_track_raises(self, tmp_path: Path) -> None:
769        """Missing cue track raises."""
770        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
771        provider = _make_provider(base_path=str(tmp_path))
772        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
773        provider._cue.parse_tracks = AsyncMock(return_value=[])  # type: ignore[method-assign]
774        item_id = make_cue_track_id(cue_item.relative_path, 1)
775        with pytest.raises(MediaNotFoundError):
776            await provider.get_track(item_id)
777
778
779class TestClassifyScanItemCue:
780    """
781    Sync-walker classification for CUE files.
782
783    Guards the edit-resync path: a CUE's previous checksum lives under synthetic
784    per-track ids in provider_mappings, never under the CUE path itself. The
785    scan reverse-derives a path-keyed map so an unchanged CUE is recognised and
786    an edited CUE forwards its prior checksum, which in turn makes the library
787    write use overwrite_existing=True.
788    """
789
790    @staticmethod
791    def _cue_item(checksum: str) -> FileSystemItem:
792        return FileSystemItem(
793            filename="album.cue",
794            relative_path="album.cue",
795            absolute_path="/music/album.cue",
796            is_dir=False,
797            checksum=checksum,
798            file_size=100,
799        )
800
801    @staticmethod
802    def _classify(
803        provider: LocalFileSystemProvider,
804        item: FileSystemItem,
805        *,
806        cue_file_checksums: dict[str, str] | None = None,
807    ) -> tuple[
808        list[tuple[FileSystemItem, str | None]],
809        list[FileSystemItem],
810        set[str],
811        set[str],
812    ]:
813        items_to_process: list[tuple[FileSystemItem, str | None]] = []
814        unchanged_cue_items: list[FileSystemItem] = []
815        cur_filenames: set[str] = set()
816        cue_stems: set[str] = set()
817        provider._classify_scan_item(
818            item,
819            file_checksums={},
820            cue_file_checksums=cue_file_checksums or {},
821            cur_filenames=cur_filenames,
822            items_to_process=items_to_process,
823            unchanged_cue_items=unchanged_cue_items,
824            cue_stems=cue_stems,
825            ignore_album_playlists=False,
826        )
827        return items_to_process, unchanged_cue_items, cur_filenames, cue_stems
828
829    def test_unchanged_cue_routes_to_unchanged_bucket(self) -> None:
830        """Matching checksum: CUE is marked present and not re-processed."""
831        provider = _make_provider()
832        cue_item = self._cue_item("checksum-v1")
833        items, unchanged, cur, stems = self._classify(
834            provider,
835            cue_item,
836            cue_file_checksums={"album.cue": "checksum-v1"},
837        )
838        assert items == []
839        assert unchanged == [cue_item]
840        assert cur == {"album.cue"}
841        assert stems == {"/music/album"}
842
843    def test_edited_cue_forwards_prior_checksum(self) -> None:
844        """Changed checksum: prior value is forwarded so downstream overwrite=True."""
845        provider = _make_provider()
846        cue_item = self._cue_item("checksum-v2")
847        items, unchanged, _, _ = self._classify(
848            provider,
849            cue_item,
850            cue_file_checksums={"album.cue": "checksum-v1"},
851        )
852        assert items == [(cue_item, "checksum-v1")]
853        assert unchanged == []
854
855    def test_new_cue_has_no_prior_checksum(self) -> None:
856        """First-time ingest: prev_checksum is None, item queued for processing."""
857        provider = _make_provider()
858        cue_item = self._cue_item("checksum-v1")
859        items, unchanged, _, _ = self._classify(
860            provider,
861            cue_item,
862            cue_file_checksums={},
863        )
864        assert items == [(cue_item, None)]
865        assert unchanged == []
866