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_performer_registers_cue_sheet_as_representative(
405        self, tmp_path: Path
406    ) -> None:
407        """
408        A per-track PERFORMER (and the album artist) register the CUE sheet, not the audio.
409
410        The companion audio file is absorbed into CUE tracks and is never itself a synced
411        item, so only the CUE sheet's own path can later be re-queued to reparse a changed
412        artist.nfo/image for these performers.
413        """
414        audio_file = tmp_path / "album.flac"
415        audio_file.write_bytes(b"")
416        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
417        provider = _make_provider(base_path=str(tmp_path))
418        tags = _make_audio_tags(duration=900.0)
419        album = Album(
420            item_id="a1",
421            provider=provider.instance_id,
422            name="Live at the BBC",
423            provider_mappings=set(),
424        )
425        self._wire_provider_for_parse(provider, album)
426
427        with patch(
428            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
429            AsyncMock(return_value=tags),
430        ):
431            await provider._cue.parse_tracks(cue_item)
432
433        # every _parse_artist call made while building the per-track performers must carry
434        # the CUE sheet's own path, never the companion audio file's
435        parse_artist_mock = cast("AsyncMock", provider._parse_artist)
436        for call in parse_artist_mock.await_args_list:
437            assert call.kwargs.get("representative_track") == cue_item.relative_path
438
439    @pytest.mark.asyncio
440    async def test_track_durations(self, tmp_path: Path) -> None:
441        """Track durations."""
442        audio_file = tmp_path / "album.flac"
443        audio_file.write_bytes(b"")
444        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
445        provider = _make_provider(base_path=str(tmp_path))
446        # total_duration = 900s (15min)
447        tags = _make_audio_tags(duration=900.0, album="Live at the BBC")
448        self._wire_provider_for_parse(provider)
449
450        with patch(
451            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
452            AsyncMock(return_value=tags),
453        ):
454            tracks = await provider._cue.parse_tracks(cue_item)
455
456        # Track 1: 00:00:00 to 04:10:40 = ~250.53s
457        # Track 2: 04:10:40 to 07:58:05 = ~227.4s
458        # Track 3: 07:58:05 to 900 = ~421.93s
459        assert tracks[0].duration == round(4 * 60 + 10 + 40 / 75)
460        assert tracks[1].duration == round((7 * 60 + 58 + 5 / 75) - (4 * 60 + 10 + 40 / 75))
461        assert tracks[2].duration == round(900.0 - (7 * 60 + 58 + 5 / 75))
462
463    @pytest.mark.asyncio
464    async def test_honors_disc_number_tag(self, tmp_path: Path) -> None:
465        """Honors disc number tag."""
466        audio_file = tmp_path / "album.flac"
467        audio_file.write_bytes(b"")
468        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
469        provider = _make_provider(base_path=str(tmp_path))
470        tags = _make_audio_tags(duration=900.0, album="Live at the BBC", disc="2")
471        self._wire_provider_for_parse(provider)
472
473        with patch(
474            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
475            AsyncMock(return_value=tags),
476        ):
477            tracks = await provider._cue.parse_tracks(cue_item)
478
479        assert all(t.disc_number == 2 for t in tracks)
480
481    @pytest.mark.asyncio
482    async def test_skips_track_missing_title(self, tmp_path: Path) -> None:
483        """Skips track missing title."""
484        audio_file = tmp_path / "album.flac"
485        audio_file.write_bytes(b"")
486        cue_text = (
487            'PERFORMER "X"\n'
488            'TITLE "Album"\n'
489            'FILE "album.flac" WAVE\n'
490            "  TRACK 01 AUDIO\n"
491            '    TITLE "Real Track"\n'
492            "    INDEX 01 00:00:00\n"
493            "  TRACK 02 AUDIO\n"
494            "    INDEX 01 02:00:00\n"  # no TITLE
495        )
496        cue_item = _make_cue_item(tmp_path, cue_text)
497        provider = _make_provider(base_path=str(tmp_path))
498        tags = _make_audio_tags(duration=300.0, album="Album")
499        self._wire_provider_for_parse(provider)
500
501        with patch(
502            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
503            AsyncMock(return_value=tags),
504        ):
505            tracks = await provider._cue.parse_tracks(cue_item)
506
507        assert len(tracks) == 1
508        assert tracks[0].name == "Real Track"
509        # a warning should have been emitted for the skipped track
510        warning_msgs = [str(c) for c in provider.logger.warning.call_args_list]  # type: ignore[attr-defined]
511        assert any("TITLE" in msg for msg in warning_msgs)
512
513    @pytest.mark.asyncio
514    async def test_track_artist_falls_back_to_album_performer(self, tmp_path: Path) -> None:
515        """Track artist falls back to album performer."""
516        audio_file = tmp_path / "album.flac"
517        audio_file.write_bytes(b"")
518        # no per-track PERFORMER, only top-level
519        cue_text = (
520            'PERFORMER "Band"\n'
521            'TITLE "Album"\n'
522            'FILE "album.flac" WAVE\n'
523            "  TRACK 01 AUDIO\n"
524            '    TITLE "T1"\n'
525            "    INDEX 01 00:00:00\n"
526        )
527        cue_item = _make_cue_item(tmp_path, cue_text)
528        provider = _make_provider(base_path=str(tmp_path))
529        tags = _make_audio_tags(duration=300.0, album="Album")
530        self._wire_provider_for_parse(provider)
531
532        with patch(
533            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
534            AsyncMock(return_value=tags),
535        ):
536            tracks = await provider._cue.parse_tracks(cue_item)
537
538        assert len(tracks) == 1
539        assert [a.name for a in tracks[0].artists] == ["Band"]
540
541    @pytest.mark.asyncio
542    async def test_track_artist_falls_back_to_unknown_when_no_performer(
543        self, tmp_path: Path
544    ) -> None:
545        """No PERFORMER at sheet or track level falls back to the [unknown] artist."""
546        audio_file = tmp_path / "album.flac"
547        audio_file.write_bytes(b"")
548        cue_text = (
549            'TITLE "Album"\n'
550            'FILE "album.flac" WAVE\n'
551            "  TRACK 01 AUDIO\n"
552            '    TITLE "T1"\n'
553            "    INDEX 01 00:00:00\n"
554        )
555        cue_item = _make_cue_item(tmp_path, cue_text)
556        provider = _make_provider(base_path=str(tmp_path))
557        tags = _make_audio_tags(duration=300.0, album="Album")
558        self._wire_provider_for_parse(provider)
559
560        with patch(
561            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
562            AsyncMock(return_value=tags),
563        ):
564            tracks = await provider._cue.parse_tracks(cue_item)
565
566        assert len(tracks) == 1
567        assert [a.name for a in tracks[0].artists] == [UNKNOWN_ARTIST]
568
569    @pytest.mark.asyncio
570    async def test_multi_line_performer_yields_multiple_artists(self, tmp_path: Path) -> None:
571        """Repeated PERFORMER lines produce one Artist each (Vorbis multi-field style)."""
572        audio_file = tmp_path / "album.flac"
573        audio_file.write_bytes(b"")
574        cue_text = (
575            'TITLE "Split"\n'
576            'FILE "album.flac" WAVE\n'
577            "  TRACK 01 AUDIO\n"
578            '    TITLE "T1"\n'
579            '    PERFORMER "AC/DC"\n'
580            '    PERFORMER "Queen"\n'
581            "    INDEX 01 00:00:00\n"
582        )
583        cue_item = _make_cue_item(tmp_path, cue_text)
584        provider = _make_provider(base_path=str(tmp_path))
585        tags = _make_audio_tags(duration=300.0, album="Split")
586        self._wire_provider_for_parse(provider)
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        # "AC/DC" is preserved intact, not split on the slash
596        assert [a.name for a in tracks[0].artists] == ["AC/DC", "Queen"]
597
598    @pytest.mark.asyncio
599    async def test_recording_and_releasetrack_mbids_mapped_distinctly(self, tmp_path: Path) -> None:
600        """REM MUSICBRAINZ_RECORDINGID → MB_RECORDING / .mbid; REM MUSICBRAINZ_TRACKID → MB_TRACK."""
601        audio_file = tmp_path / "album.flac"
602        audio_file.write_bytes(b"")
603        recording_mbid = "11111111-1111-1111-1111-111111111111"
604        releasetrack_mbid = "22222222-2222-2222-2222-222222222222"
605        cue_text = (
606            'TITLE "Album"\n'
607            'FILE "album.flac" WAVE\n'
608            "  TRACK 01 AUDIO\n"
609            '    TITLE "T1"\n'
610            f"    REM MUSICBRAINZ_RECORDINGID {recording_mbid}\n"
611            f"    REM MUSICBRAINZ_TRACKID {releasetrack_mbid}\n"
612            "    INDEX 01 00:00:00\n"
613        )
614        cue_item = _make_cue_item(tmp_path, cue_text)
615        provider = _make_provider(base_path=str(tmp_path))
616        tags = _make_audio_tags(duration=300.0, album="Album")
617        self._wire_provider_for_parse(provider)
618
619        with patch(
620            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
621            AsyncMock(return_value=tags),
622        ):
623            tracks = await provider._cue.parse_tracks(cue_item)
624
625        assert len(tracks) == 1
626        assert (ExternalID.MB_RECORDING, recording_mbid) in tracks[0].external_ids
627        assert (ExternalID.MB_TRACK, releasetrack_mbid) in tracks[0].external_ids
628        assert tracks[0].mbid == recording_mbid
629
630    @pytest.mark.asyncio
631    async def test_aligned_track_artist_metadata(self, tmp_path: Path) -> None:
632        """REM ARTISTSORT / REM MUSICBRAINZ_ARTISTID align by index with PERFORMER."""
633        audio_file = tmp_path / "album.flac"
634        audio_file.write_bytes(b"")
635        cue_text = (
636            'TITLE "Album"\n'
637            'FILE "album.flac" WAVE\n'
638            "  TRACK 01 AUDIO\n"
639            '    TITLE "T1"\n'
640            '    PERFORMER "First Artist"\n'
641            '    PERFORMER "Second Artist"\n'
642            '    REM ARTISTSORT "Artist, First"\n'
643            '    REM ARTISTSORT "Artist, Second"\n'
644            "    REM MUSICBRAINZ_ARTISTID 11111111-1111-1111-1111-111111111111\n"
645            "    REM MUSICBRAINZ_ARTISTID 22222222-2222-2222-2222-222222222222\n"
646            "    INDEX 01 00:00:00\n"
647        )
648        cue_item = _make_cue_item(tmp_path, cue_text)
649        provider = _make_provider(base_path=str(tmp_path))
650        tags = _make_audio_tags(duration=300.0, album="Album")
651        self._wire_provider_for_parse(provider)
652        # override _parse_artist to capture the sort_name/mbid args passed per artist
653        captured: list[dict[str, str | None]] = []
654
655        async def _capture(
656            name: str, sort_name: str | None = None, mbid: str | None = None, **_k: object
657        ) -> Artist:
658            captured.append({"name": name, "sort_name": sort_name, "mbid": mbid})
659            return Artist(
660                item_id=name,
661                provider=provider.instance_id,
662                name=name,
663                sort_name=sort_name,
664                provider_mappings=set(),
665            )
666
667        provider._parse_artist = AsyncMock(side_effect=_capture)  # type: ignore[method-assign]
668
669        with patch(
670            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
671            AsyncMock(return_value=tags),
672        ):
673            tracks = await provider._cue.parse_tracks(cue_item)
674
675        assert len(tracks) == 1
676        assert captured == [
677            {
678                "name": "First Artist",
679                "sort_name": "Artist, First",
680                "mbid": "11111111-1111-1111-1111-111111111111",
681            },
682            {
683                "name": "Second Artist",
684                "sort_name": "Artist, Second",
685                "mbid": "22222222-2222-2222-2222-222222222222",
686            },
687        ]
688
689    @pytest.mark.asyncio
690    async def test_track_level_descriptive_fields(self, tmp_path: Path) -> None:
691        """REM COPYRIGHT / GROUPING / COMMENT / ITUNESADVISORY / TITLESORT populate track metadata."""
692        audio_file = tmp_path / "album.flac"
693        audio_file.write_bytes(b"")
694        cue_text = (
695            'TITLE "Album"\n'
696            'FILE "album.flac" WAVE\n'
697            "  TRACK 01 AUDIO\n"
698            '    TITLE "Song, The"\n'
699            '    REM TITLESORT "Song, The"\n'
700            '    REM COPYRIGHT "(c) 2024 Label"\n'
701            '    REM GROUPING "Movement I"\n'
702            '    REM COMMENT "Live at Wembley"\n'
703            "    REM ITUNESADVISORY 1\n"
704            "    INDEX 01 00:00:00\n"
705        )
706        cue_item = _make_cue_item(tmp_path, cue_text)
707        provider = _make_provider(base_path=str(tmp_path))
708        tags = _make_audio_tags(duration=300.0, album="Album")
709        self._wire_provider_for_parse(provider)
710
711        with patch(
712            "music_assistant.providers.filesystem_local.cue.async_parse_tags",
713            AsyncMock(return_value=tags),
714        ):
715            tracks = await provider._cue.parse_tracks(cue_item)
716
717        assert len(tracks) == 1
718        track = tracks[0]
719        assert track.sort_name == "Song, The"
720        assert track.metadata.copyright == "(c) 2024 Label"
721        assert track.metadata.grouping == "Movement I"
722        assert track.metadata.description == "Live at Wembley"
723        assert track.metadata.explicit is True
724
725
726class TestGetStreamDetailsForCueTrack:
727    """Tests for _get_stream_details_for_cue_track."""
728
729    @pytest.mark.asyncio
730    async def test_invalid_id_raises(self) -> None:
731        """Invalid id raises."""
732        provider = _make_provider()
733        with pytest.raises(InvalidDataError):
734            await provider._cue.get_stream_details("not_a_cue_id.flac")
735
736    @pytest.mark.asyncio
737    async def test_not_in_library_raises(self, tmp_path: Path) -> None:
738        """Track not in library raises."""
739        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
740        provider = _make_provider(base_path=str(tmp_path))
741        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
742        provider.mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)  # type: ignore[method-assign]
743        item_id = make_cue_track_id(cue_item.relative_path, 1)
744        with pytest.raises(MediaNotFoundError):
745            await provider._cue.get_stream_details(item_id)
746
747    @pytest.mark.asyncio
748    async def test_missing_audio_raises(self, tmp_path: Path) -> None:
749        """Missing audio raises."""
750        cue_item = _make_cue_item(
751            tmp_path,
752            'FILE "missing.flac" WAVE\n  TRACK 01 AUDIO\n    TITLE "x"\n    INDEX 01 00:00:00\n',
753        )
754        provider = _make_provider(base_path=str(tmp_path))
755        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
756        item_id = make_cue_track_id(cue_item.relative_path, 1)
757        _stub_library_track(provider, item_id)
758        with pytest.raises(MediaNotFoundError):
759            await provider._cue.get_stream_details(item_id)
760
761    @pytest.mark.asyncio
762    async def test_unknown_track_number_raises(self, tmp_path: Path) -> None:
763        """Unknown track number raises."""
764        audio_file = tmp_path / "album.flac"
765        audio_file.write_bytes(b"")
766        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
767        provider = _make_provider(base_path=str(tmp_path))
768        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
769        # request track 99 which isn't in the CUE
770        item_id = make_cue_track_id(cue_item.relative_path, 99)
771        _stub_library_track(provider, item_id)
772        with pytest.raises(MediaNotFoundError):
773            await provider._cue.get_stream_details(item_id)
774
775    @pytest.mark.asyncio
776    async def test_builds_streamdetails_with_offset_and_duration(self, tmp_path: Path) -> None:
777        """Builds streamdetails with offset and duration."""
778        audio_file = tmp_path / "album.flac"
779        audio_file.write_bytes(b"")
780        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
781        provider = _make_provider(base_path=str(tmp_path))
782        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
783        item_id = make_cue_track_id(cue_item.relative_path, 2)
784        _stub_library_track(provider, item_id, duration=228)
785
786        details = await provider._cue.get_stream_details(item_id)
787
788        assert isinstance(details, StreamDetails)
789        assert details.item_id == item_id
790        assert details.can_seek is True
791        assert details.allow_seek is True
792        assert details.audio_format.content_type == ContentType.PCM_F32LE
793        assert details.duration == 228
794        assert details.data is not None
795        assert details.data["audio_relative_path"] == "album.flac"
796        # Track 2 starts at 04:10:40 = 250.533...
797        expected_start = 4 * 60 + 10 + 40 / 75
798        assert abs(details.data["start_seconds"] - expected_start) < 0.001
799
800
801class TestProcessDeletionsCueBranch:
802    """Tests for _process_deletions routing of CUE-derived track ids."""
803
804    @pytest.mark.asyncio
805    async def test_cue_track_id_routed_to_track_controller(self) -> None:
806        """Cue track id routed to track controller."""
807        provider = _make_provider()
808        controller = MagicMock()
809        controller.get_library_item_by_prov_id = AsyncMock(return_value=None)
810        provider.mass.music.get_controller = MagicMock(return_value=controller)  # type: ignore[method-assign]
811
812        cue_track_id = make_cue_track_id("artist/album.cue", 3)
813        await provider._process_deletions({cue_track_id})
814
815        # must have consulted a controller (the track controller)
816        assert provider.mass.music.get_controller.called
817
818
819class TestGetTrackCueBranch:
820    """Test get_track's CUE-id branch."""
821
822    @pytest.mark.asyncio
823    async def test_returns_matching_cue_track(self, tmp_path: Path) -> None:
824        """Returns matching cue track."""
825        audio_file = tmp_path / "album.flac"
826        audio_file.write_bytes(b"")
827        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
828        provider = _make_provider(base_path=str(tmp_path))
829        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
830
831        # mock _parse_cue_tracks to return three synthetic tracks
832        def fake_track(num: int) -> Track:
833            return Track(
834                item_id=make_cue_track_id(cue_item.relative_path, num),
835                provider=provider.instance_id,
836                name=f"Track {num}",
837                provider_mappings=set(),
838            )
839
840        provider._cue.parse_tracks = AsyncMock(  # type: ignore[method-assign]
841            return_value=[fake_track(1), fake_track(2), fake_track(3)]
842        )
843        item_id = make_cue_track_id(cue_item.relative_path, 2)
844        track = await provider.get_track(item_id)
845        assert track.item_id == item_id
846        assert track.name == "Track 2"
847
848    @pytest.mark.asyncio
849    async def test_missing_cue_track_raises(self, tmp_path: Path) -> None:
850        """Missing cue track raises."""
851        cue_item = _make_cue_item(tmp_path, SAMPLE_CUE)
852        provider = _make_provider(base_path=str(tmp_path))
853        provider.resolve = AsyncMock(return_value=cue_item)  # type: ignore[method-assign]
854        provider._cue.parse_tracks = AsyncMock(return_value=[])  # type: ignore[method-assign]
855        item_id = make_cue_track_id(cue_item.relative_path, 1)
856        with pytest.raises(MediaNotFoundError):
857            await provider.get_track(item_id)
858
859
860class TestClassifyScanItemCue:
861    """
862    Sync-walker classification for CUE files.
863
864    Guards the edit-resync path: a CUE's previous checksum lives under synthetic
865    per-track ids in provider_mappings, never under the CUE path itself. The
866    scan reverse-derives a path-keyed map so an unchanged CUE is recognised and
867    an edited CUE forwards its prior checksum, which in turn makes the library
868    write use overwrite_existing=True.
869    """
870
871    @staticmethod
872    def _cue_item(checksum: str) -> FileSystemItem:
873        return FileSystemItem(
874            filename="album.cue",
875            relative_path="album.cue",
876            absolute_path="/music/album.cue",
877            is_dir=False,
878            checksum=checksum,
879            file_size=100,
880        )
881
882    @staticmethod
883    def _classify(
884        provider: LocalFileSystemProvider,
885        item: FileSystemItem,
886        *,
887        cue_file_checksums: dict[str, set[str]] | None = None,
888    ) -> tuple[
889        list[tuple[FileSystemItem, str | None]],
890        list[FileSystemItem],
891        set[str],
892        set[str],
893    ]:
894        items_to_process: list[tuple[FileSystemItem, str | None]] = []
895        unchanged_cue_items: list[FileSystemItem] = []
896        cur_filenames: set[str] = set()
897        cue_stems: set[str] = set()
898        provider._classify_scan_item(
899            item,
900            file_checksums={},
901            cue_file_checksums=cue_file_checksums or {},
902            cur_filenames=cur_filenames,
903            items_to_process=items_to_process,
904            unchanged_cue_items=unchanged_cue_items,
905            cue_stems=cue_stems,
906            ignore_album_playlists=False,
907            metadata_files=[],
908        )
909        return items_to_process, unchanged_cue_items, cur_filenames, cue_stems
910
911    def test_unchanged_cue_routes_to_unchanged_bucket(self) -> None:
912        """Matching checksum: CUE is marked present and not re-processed."""
913        provider = _make_provider()
914        cue_item = self._cue_item("checksum-v1")
915        items, unchanged, cur, stems = self._classify(
916            provider,
917            cue_item,
918            cue_file_checksums={"album.cue": {cue_metadata_checksum("checksum-v1")}},
919        )
920        assert items == []
921        assert unchanged == [cue_item]
922        assert cur == {"album.cue"}
923        assert stems == {"/music/album"}
924
925    def test_edited_cue_forwards_prior_checksum(self) -> None:
926        """Changed checksum: prior value is forwarded so downstream overwrite=True."""
927        provider = _make_provider()
928        cue_item = self._cue_item("checksum-v2")
929        items, unchanged, _, _ = self._classify(
930            provider,
931            cue_item,
932            cue_file_checksums={"album.cue": {cue_metadata_checksum("checksum-v1")}},
933        )
934        assert items == [(cue_item, cue_metadata_checksum("checksum-v1"))]
935        assert unchanged == []
936
937    def test_legacy_checksum_forces_metadata_refresh(self) -> None:
938        """An unversioned mapping is reprocessed even when the file is unchanged."""
939        provider = _make_provider()
940        cue_item = self._cue_item("checksum-v1")
941        items, unchanged, _, _ = self._classify(
942            provider,
943            cue_item,
944            cue_file_checksums={"album.cue": {"checksum-v1"}},
945        )
946        assert items == [(cue_item, "checksum-v1")]
947        assert unchanged == []
948
949    def test_mixed_checksums_force_metadata_refresh(self) -> None:
950        """A partially refreshed CUE is reprocessed until every track is current."""
951        provider = _make_provider()
952        cue_item = self._cue_item("checksum-v1")
953        previous_checksums = {
954            "checksum-v1",
955            cue_metadata_checksum("checksum-v1"),
956        }
957        items, unchanged, _, _ = self._classify(
958            provider,
959            cue_item,
960            cue_file_checksums={"album.cue": previous_checksums},
961        )
962        assert items == [(cue_item, min(previous_checksums))]
963        assert unchanged == []
964
965    def test_new_cue_has_no_prior_checksum(self) -> None:
966        """First-time ingest: prev_checksum is None, item queued for processing."""
967        provider = _make_provider()
968        cue_item = self._cue_item("checksum-v1")
969        items, unchanged, _, _ = self._classify(
970            provider,
971            cue_item,
972            cue_file_checksums={},
973        )
974        assert items == [(cue_item, None)]
975        assert unchanged == []
976