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