music-assistant-server

1"""Tests for the AcoustidLookupProvider."""
2
3from __future__ import annotations
4
5import sys
6from types import ModuleType
7from typing import Any, cast
8from unittest.mock import AsyncMock, MagicMock
9
10import aiohttp
11import pytest
12from music_assistant_models.enums import MediaType, StreamType
13from music_assistant_models.errors import UnsupportedSystemError
14
15from music_assistant.constants import CONF_LOG_LEVEL
16from music_assistant.helpers.datetime import utc_timestamp
17from music_assistant.models.audio_analysis import AudioAnalysisData
18from music_assistant.models.audio_analysis_provider import AnalysisSessionData
19from music_assistant.providers.acoustid_lookup.provider import (
20    CONF_ANALYSE_STREAMING,
21    CONF_API_KEY,
22    CONF_MIN_SCORE,
23    CONF_WRITE_TAGS_BACK,
24    NO_MATCH_ANALYSIS_VERSION,
25    NO_MATCH_RETRY_DAYS,
26    AcoustidLookupProvider,
27    _AcoustidSessionData,
28    _parse_response,
29)
30
31# ---------------------------------------------------------------------------
32# Fixtures / helpers
33# ---------------------------------------------------------------------------
34
35
36def _make_provider(
37    *,
38    api_key: str | None = "TESTKEY",
39    min_score: float = 0.85,
40    write_tags_back: bool = False,
41    analyse_streaming: bool = True,
42) -> AcoustidLookupProvider:
43    """Build a provider with a mocked MusicAssistant infrastructure."""
44    mass = MagicMock()
45
46    def _default_get_provider(provider_id: Any, **_kwargs: Any) -> Any:
47        # Source-provider lookup feeds the write_access gate in post_analysis; default to
48        # a writeable filesystem-like source so tag-write paths are exercised by default.
49        if provider_id == "filesystem_local_test":
50            return MagicMock(write_access=True)
51        return None
52
53    mass.get_provider = MagicMock(side_effect=_default_get_provider)
54    mass.streams.audio_analysis.get_audio_analysis_version = AsyncMock(return_value=None)
55    mass.streams.audio_analysis.get_audio_analysis = AsyncMock(return_value=None)
56    mass.streams.audio_analysis.set_audio_analysis = AsyncMock()
57    mass.music.tracks.set_identifiers = AsyncMock()
58    mass.music.albums.set_release_group = AsyncMock()
59    mass.streams.audio_analysis.get_extra_data_for_album_tracks = AsyncMock(return_value=[])
60    mass.music.tracks.get_library_item_by_prov_id = AsyncMock(return_value=None)
61    mass.music.albums.get_library_item = AsyncMock(return_value=None)
62    mass.music.albums.get_library_album_tracks = AsyncMock(return_value=[])
63    mass.cache.get = AsyncMock(return_value=None)
64    mass.cache.set = AsyncMock()
65    mass.http_session = MagicMock()
66
67    manifest = MagicMock()
68    manifest.domain = "acoustid_lookup"
69
70    config = MagicMock()
71    config.instance_id = "acoustid_lookup_test"
72    config.values = {}
73    config_lookup: dict[str, Any] = {
74        CONF_LOG_LEVEL: "GLOBAL",
75        CONF_API_KEY: api_key,
76        CONF_MIN_SCORE: min_score,
77        CONF_WRITE_TAGS_BACK: write_tags_back,
78        CONF_ANALYSE_STREAMING: analyse_streaming,
79    }
80    config.get_value = MagicMock(side_effect=lambda key: config_lookup.get(key, "GLOBAL"))
81
82    return AcoustidLookupProvider(mass, manifest, config, set())
83
84
85def _make_streamdetails(
86    *,
87    stream_type: StreamType = StreamType.LOCAL_FILE,
88    media_type: MediaType = MediaType.TRACK,
89    path: str | None = "/music/track.flac",
90    duration: int | None = 180,
91) -> MagicMock:
92    sd = MagicMock()
93    sd.item_id = "track-1"
94    sd.provider = "filesystem_local_test"
95    sd.uri = "library://track/track-1"
96    sd.media_type = media_type
97    sd.stream_type = stream_type
98    sd.path = path
99    sd.duration = duration
100    return sd
101
102
103def _make_audio_format(
104    *, bit_depth: int = 16, sample_rate: int = 44100, channels: int = 2
105) -> MagicMock:
106    fmt = MagicMock()
107    fmt.bit_depth = bit_depth
108    fmt.sample_rate = sample_rate
109    fmt.channels = channels
110    return fmt
111
112
113class _FakeFingerprinter:
114    """Stand-in for acoustid.chromaprint.Fingerprinter."""
115
116    def __init__(self) -> None:
117        self.start_args: tuple[int, int] | None = None
118        self.fed: list[bytes] = []
119        self.finished = False
120
121    def start(self, sample_rate: int, channels: int) -> None:
122        self.start_args = (sample_rate, channels)
123
124    def feed(self, data: bytes) -> None:
125        self.fed.append(bytes(data))
126
127    def finish(self) -> bytes:
128        self.finished = True
129        return b"FAKEFINGERPRINT"
130
131
132def _install_fake_chromaprint(monkeypatch: pytest.MonkeyPatch, fp: _FakeFingerprinter) -> None:
133    """Patch _create_fingerprinter to return the given fake, started with the session's format."""
134
135    def fake_create(
136        _self: AcoustidLookupProvider, sample_rate: int, channels: int
137    ) -> _FakeFingerprinter:
138        fp.start(int(sample_rate), int(channels))
139        return fp
140
141    monkeypatch.setattr(AcoustidLookupProvider, "_create_fingerprinter", fake_create)
142
143
144def _install_unidentified_track(provider: AcoustidLookupProvider) -> None:
145    """Make the provider's library lookup return a track with no MBID or ISRC."""
146    track = MagicMock(mbid=None)
147    track.get_external_id.return_value = None
148    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
149        return_value=track
150    )
151
152
153class _FakeFingerprintError(Exception):
154    """Stand-in for chromaprint.FingerprintError."""
155
156
157def _install_chromaprint_module(monkeypatch: pytest.MonkeyPatch, *, failing_call: str) -> None:
158    """
159    Inject a stand-in ``chromaprint`` module so the real _create_fingerprinter runs.
160
161    :param failing_call: Fingerprinter method ("start", "feed" or "finish") that
162        raises, letting a test drive one native error path.
163    """
164
165    class _Fingerprinter:
166        def start(self, sample_rate: int, channels: int) -> None:
167            if failing_call == "start":
168                raise _FakeFingerprintError("start failed")
169
170        def feed(self, data: bytes) -> None:
171            if failing_call == "feed":
172                raise _FakeFingerprintError("feed failed")
173
174        def finish(self) -> bytes:
175            if failing_call == "finish":
176                raise _FakeFingerprintError("finish failed")
177            return b"FAKEFINGERPRINT"
178
179    module = ModuleType("chromaprint")
180    module.FingerprintError = _FakeFingerprintError  # type: ignore[attr-defined]
181    module.Fingerprinter = _Fingerprinter  # type: ignore[attr-defined]
182    monkeypatch.setitem(sys.modules, "chromaprint", module)
183
184
185def _install_lookup_response(
186    monkeypatch: pytest.MonkeyPatch, response: dict[str, Any] | None
187) -> None:
188    """Patch the cached/throttled _lookup to return a fixed dict (or None)."""
189
190    async def fake_lookup(
191        _self: AcoustidLookupProvider, *_args: Any, **_kwargs: Any
192    ) -> dict[str, Any] | None:
193        return response
194
195    monkeypatch.setattr(AcoustidLookupProvider, "_lookup", fake_lookup)
196
197
198def _make_library_album(
199    *,
200    name: str = "Silver Thunderbird",
201    existing_rg: str | None = None,
202) -> MagicMock:
203    """Build a library Album mock with get/add_external_id wiring."""
204    album = MagicMock()
205    album.name = name
206    album.item_id = "42"
207    storage: dict[str, str] = {}
208    if existing_rg:
209        storage["musicbrainz_releasegroupid"] = existing_rg
210
211    def get_ext(key: Any) -> str | None:
212        return storage.get(getattr(key, "value", key))
213
214    def add_ext(key: Any, value: str) -> None:
215        storage[getattr(key, "value", key)] = value
216
217    album.get_external_id.side_effect = get_ext
218    album.add_external_id.side_effect = add_ext
219    return album
220
221
222def _make_album_tracks(count: int, provider_instance: str = "filesystem_local_test") -> list[Any]:
223    """Build N Track mocks each with one provider mapping in our provider."""
224    tracks = []
225    for i in range(count):
226        pm = MagicMock()
227        pm.provider_instance = provider_instance
228        pm.provider_domain = "filesystem_local"
229        pm.item_id = f"native-{i}"
230        t = MagicMock()
231        t.provider_mappings = [pm]
232        tracks.append(t)
233    return tracks
234
235
236def _wire_album_for_consensus(
237    provider: AcoustidLookupProvider,
238    *,
239    album: MagicMock,
240    album_tracks: list[Any],
241    extras: list[dict[str, Any]],
242) -> None:
243    """Glue an album + tracks + AcoustID extras into the provider's mass mock."""
244    library_track = MagicMock()
245    library_track.album = MagicMock(item_id="42")
246    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
247        return_value=library_track
248    )
249    cast("MagicMock", provider.mass.music.albums).get_library_item = AsyncMock(return_value=album)
250    cast("MagicMock", provider.mass.music.albums).get_library_album_tracks = AsyncMock(
251        return_value=album_tracks
252    )
253    cast(
254        "MagicMock", provider.mass.streams.audio_analysis
255    ).get_extra_data_for_album_tracks = AsyncMock(return_value=extras)
256
257
258def _rg(
259    rg_id: str, title: str = "Silver Thunderbird", primary_type: str = "Album"
260) -> dict[str, Any]:
261    """Shortcut for a release-group entry inside ``extras``."""
262    return {"id": rg_id, "title": title, "primary_type": primary_type}
263
264
265# ---------------------------------------------------------------------------
266# Provider core
267# ---------------------------------------------------------------------------
268
269
270@pytest.mark.asyncio
271@pytest.mark.parametrize(
272    (
273        "provider_kwargs",
274        "streamdetails_kwargs",
275        "track_mbid",
276        "track_isrc",
277        "track_in_library",
278        "expected",
279    ),
280    [
281        pytest.param({}, {}, "0000-mbid", None, True, False, id="mbid_already_present"),
282        pytest.param({}, {}, None, "USRC17607839", True, False, id="isrc_already_present"),
283        pytest.param({}, {}, None, None, False, False, id="no_library_row"),
284        pytest.param(
285            {},
286            {"media_type": MediaType.PODCAST_EPISODE},
287            None,
288            None,
289            False,
290            False,
291            id="non_track_media",
292        ),
293        pytest.param(
294            {"analyse_streaming": False},
295            {"stream_type": StreamType.HTTP},
296            None,
297            None,
298            True,
299            False,
300            id="streaming_toggle_off",
301        ),
302        pytest.param(
303            {"analyse_streaming": True},
304            {"stream_type": StreamType.HTTP},
305            None,
306            None,
307            True,
308            True,
309            id="streaming_toggle_on",
310        ),
311        pytest.param({}, {}, None, None, True, True, id="local_file_mbid_missing"),
312        # No user-supplied key still proceeds — the shared AcoustID key is used.
313        pytest.param({"api_key": None}, {}, None, None, True, True, id="no_user_api_key"),
314    ],
315)
316async def test_start_analysis_gates(
317    monkeypatch: pytest.MonkeyPatch,
318    *,
319    provider_kwargs: dict[str, Any],
320    streamdetails_kwargs: dict[str, Any],
321    track_mbid: str | None,
322    track_isrc: str | None,
323    track_in_library: bool,
324    expected: bool,
325) -> None:
326    """start_analysis accepts the session or skips for each precondition."""
327    provider = _make_provider(**provider_kwargs)
328    if track_in_library:
329        track = MagicMock()
330        track.mbid = track_mbid
331        track.get_external_id.return_value = track_isrc
332        cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
333            return_value=track
334        )
335    _install_fake_chromaprint(monkeypatch, _FakeFingerprinter())
336
337    accepted = await provider.start_analysis(
338        "session", _make_streamdetails(**streamdetails_kwargs), _make_audio_format()
339    )
340
341    assert accepted is expected
342
343
344@pytest.mark.asyncio
345@pytest.mark.parametrize(
346    ("track_mbid", "track_isrc", "expected_extra"),
347    [
348        pytest.param(
349            "0000-mbid", None, {"source": "existing_tags", "mbid": "0000-mbid"}, id="mbid_only"
350        ),
351        pytest.param(
352            None,
353            "USRC17607839",
354            {"source": "existing_tags", "isrc": "USRC17607839"},
355            id="isrc_only",
356        ),
357        pytest.param(
358            "0000-mbid",
359            "USRC17607839",
360            {"source": "existing_tags", "mbid": "0000-mbid", "isrc": "USRC17607839"},
361            id="mbid_and_isrc",
362        ),
363    ],
364)
365async def test_start_analysis_records_existing_identifiers(
366    monkeypatch: pytest.MonkeyPatch,
367    *,
368    track_mbid: str | None,
369    track_isrc: str | None,
370    expected_extra: dict[str, Any],
371) -> None:
372    """An already-identified track is recorded as analyzed (no fingerprinting, no retry)."""
373    provider = _make_provider()
374    track = MagicMock()
375    track.mbid = track_mbid
376    track.get_external_id.return_value = track_isrc
377    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
378        return_value=track
379    )
380    _install_fake_chromaprint(monkeypatch, _FakeFingerprinter())
381    set_aa = cast("AsyncMock", provider.mass.streams.audio_analysis.set_audio_analysis)
382
383    accepted = await provider.start_analysis("session", _make_streamdetails(), _make_audio_format())
384
385    # session is declined (no streaming) but a result row is recorded
386    assert accepted is False
387    set_aa.assert_awaited_once()
388    assert set_aa.await_args is not None
389    kwargs = set_aa.await_args.kwargs
390    assert kwargs["aa_provider_domain"] == provider.domain
391    assert kwargs["item_id"] == "track-1"
392    analysis = kwargs["analysis"]
393    assert analysis.extra_data == expected_extra
394    # permanent result — never re-collected by the background scan
395    assert "retry_after" not in analysis.extra_data
396
397
398@pytest.mark.asyncio
399async def test_start_analysis_skips_within_no_match_cooldown(
400    monkeypatch: pytest.MonkeyPatch,
401) -> None:
402    """A track whose prior no-match result is still inside its cooldown is declined."""
403    provider = _make_provider()
404    track = MagicMock(mbid=None)
405    track.get_external_id.return_value = None
406    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
407        return_value=track
408    )
409    cast("MagicMock", provider.mass.streams.audio_analysis).get_audio_analysis = AsyncMock(
410        return_value=AudioAnalysisData(extra_data={"retry_after": int(utc_timestamp()) + 3600})
411    )
412    fp = _FakeFingerprinter()
413    _install_fake_chromaprint(monkeypatch, fp)
414
415    accepted = await provider.start_analysis("session", _make_streamdetails(), _make_audio_format())
416
417    assert accepted is False
418    # declined before any fingerprinting work
419    assert fp.start_args is None
420
421
422@pytest.mark.asyncio
423async def test_start_analysis_skips_multichannel(monkeypatch: pytest.MonkeyPatch) -> None:
424    """
425    A multichannel (e.g. 5.1) file is declined before any fingerprinting work.
426
427    Feeding chromaprint multichannel audio trips a C-level assertion that aborts the
428    process, so the session must be refused up-front rather than risk the crash.
429    """
430    provider = _make_provider()
431    track = MagicMock(mbid=None)
432    track.get_external_id.return_value = None
433    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
434        return_value=track
435    )
436    fp = _FakeFingerprinter()
437    _install_fake_chromaprint(monkeypatch, fp)
438
439    accepted = await provider.start_analysis(
440        "session", _make_streamdetails(), _make_audio_format(channels=6)
441    )
442
443    assert accepted is False
444    # declined before the fingerprinter was ever started
445    assert fp.start_args is None
446
447
448@pytest.mark.asyncio
449async def test_start_analysis_retries_after_cooldown(monkeypatch: pytest.MonkeyPatch) -> None:
450    """Once the cooldown has elapsed, the track is accepted for a fresh lookup."""
451    provider = _make_provider()
452    track = MagicMock(mbid=None)
453    track.get_external_id.return_value = None
454    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
455        return_value=track
456    )
457    cast("MagicMock", provider.mass.streams.audio_analysis).get_audio_analysis = AsyncMock(
458        return_value=AudioAnalysisData(extra_data={"retry_after": int(utc_timestamp()) - 3600})
459    )
460    _install_fake_chromaprint(monkeypatch, _FakeFingerprinter())
461
462    accepted = await provider.start_analysis("session", _make_streamdetails(), _make_audio_format())
463
464    assert accepted is True
465
466
467@pytest.mark.asyncio
468async def test_finalize_happy_path(monkeypatch: pytest.MonkeyPatch) -> None:
469    """A high-score match yields mbid + acoustid + candidates + release_groups."""
470    provider = _make_provider()
471    _install_fake_chromaprint(monkeypatch, _FakeFingerprinter())
472    track = MagicMock(mbid=None)
473    track.get_external_id.return_value = None
474    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
475        return_value=track
476    )
477
478    session_id = "session-final"
479    await provider._start_analysis(session_id, _make_streamdetails(), _make_audio_format())
480    provider._sessions[session_id] = AnalysisSessionData(
481        streamdetails=_make_streamdetails(),
482        audio_format=_make_audio_format(),
483    )
484    provider._data[session_id].pcm_seconds_fed = 30.0
485
486    _install_lookup_response(
487        monkeypatch,
488        {
489            "status": "ok",
490            "results": [
491                {
492                    "id": "acoustid-1",
493                    "score": 0.97,
494                    "recordings": [
495                        {
496                            "id": "mbid-rich",
497                            "title": "Song",
498                            "artists": [{}],
499                            "releases": [{"id": "rel-a"}],
500                            "releasegroups": [
501                                {"id": "rg-a", "title": "Album", "type": "Album"},
502                            ],
503                        },
504                    ],
505                },
506            ],
507        },
508    )
509
510    result = await provider._finalize(session_id)
511
512    assert result is not None
513    assert result.extra_data is not None
514    assert result.extra_data["mbid"] == "mbid-rich"
515    assert result.extra_data["acoustid"] == "acoustid-1"
516    assert result.extra_data["match_score"] == pytest.approx(0.97)
517    assert result.extra_data["release_groups"] == [
518        {
519            "id": "rg-a",
520            "title": "Album",
521            "primary_type": "Album",
522            "secondary_types": [],
523            "artists": [],
524        }
525    ]
526    assert result.extra_data["candidates"][0]["acoustid"] == "acoustid-1"
527
528
529@pytest.mark.asyncio
530async def test_finalize_rejects_low_score(monkeypatch: pytest.MonkeyPatch) -> None:
531    """A best-result score below the threshold yields a retryable no-match marker."""
532    provider = _make_provider(min_score=0.85)
533    _install_fake_chromaprint(monkeypatch, _FakeFingerprinter())
534    track = MagicMock(mbid=None)
535    track.get_external_id.return_value = None
536    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
537        return_value=track
538    )
539
540    session_id = "session-lowscore"
541    await provider._start_analysis(session_id, _make_streamdetails(), _make_audio_format())
542    provider._sessions[session_id] = AnalysisSessionData(
543        streamdetails=_make_streamdetails(),
544        audio_format=_make_audio_format(),
545    )
546    provider._data[session_id].pcm_seconds_fed = 10.0
547
548    _install_lookup_response(
549        monkeypatch,
550        {
551            "status": "ok",
552            "results": [
553                {
554                    "id": "acoustid-low",
555                    "score": 0.5,
556                    "recordings": [{"id": "mbid-low", "title": "Song"}],
557                }
558            ],
559        },
560    )
561
562    # _finalize persists a no-match result itself and returns None so the base class
563    # does not overwrite it at the current analysis_version
564    assert await provider._finalize(session_id) is None
565    set_aa = cast("AsyncMock", provider.mass.streams.audio_analysis.set_audio_analysis)
566    set_aa.assert_awaited_once()
567    assert set_aa.await_args is not None
568    kwargs = set_aa.await_args.kwargs
569    # stored below the current version so it is re-offered, and gated on a future retry_after
570    assert kwargs["analysis_version"] == NO_MATCH_ANALYSIS_VERSION
571    now = int(utc_timestamp())
572    retry_after = kwargs["analysis"].extra_data["retry_after"]
573    assert now < retry_after <= now + NO_MATCH_RETRY_DAYS * 86400 + 5
574
575
576# ---------------------------------------------------------------------------
577# Chromaprint binding
578# ---------------------------------------------------------------------------
579
580
581@pytest.mark.asyncio
582async def test_async_init_imports_the_native_binding(monkeypatch: pytest.MonkeyPatch) -> None:
583    """A missing libchromaprint surfaces as a provider load failure, not a per-track error."""
584    provider = _make_provider()
585    imported: list[str] = []
586
587    async def fake_import(name: str, _package: str | None = None) -> ModuleType:
588        imported.append(name)
589        raise ImportError("couldn't find libchromaprint")
590
591    monkeypatch.setattr(
592        "music_assistant.providers.acoustid_lookup.provider.import_module_in_thread", fake_import
593    )
594
595    # UnsupportedSystemError marks the failure as permanent, so the loader reports it
596    # to the user instead of retrying a library that will not appear on its own.
597    with pytest.raises(UnsupportedSystemError):
598        await provider.handle_async_init()
599    assert imported == ["chromaprint"]
600
601
602@pytest.mark.asyncio
603async def test_fingerprinter_start_failure_declines_session(
604    monkeypatch: pytest.MonkeyPatch,
605) -> None:
606    """A chromaprint error while starting the fingerprinter declines the session."""
607    _install_chromaprint_module(monkeypatch, failing_call="start")
608    provider = _make_provider()
609    _install_unidentified_track(provider)
610
611    accepted = await provider._start_analysis(
612        "session", _make_streamdetails(), _make_audio_format()
613    )
614
615    assert accepted is False
616
617
618@pytest.mark.asyncio
619async def test_chromaprint_error_while_feeding_is_caught(monkeypatch: pytest.MonkeyPatch) -> None:
620    """A chromaprint error from feed() marks the session errored instead of propagating."""
621    _install_chromaprint_module(monkeypatch, failing_call="feed")
622    provider = _make_provider()
623    _install_unidentified_track(provider)
624    session_id = "session-feed"
625    assert await provider._start_analysis(session_id, _make_streamdetails(), _make_audio_format())
626
627    await provider.process_pcm_chunk(session_id, b"\x00\x01" * 100)
628
629    assert provider._data[session_id].error is not None
630
631
632@pytest.mark.asyncio
633async def test_chromaprint_error_while_finishing_is_caught(
634    monkeypatch: pytest.MonkeyPatch,
635) -> None:
636    """A chromaprint error from finish() yields no result instead of propagating."""
637    _install_chromaprint_module(monkeypatch, failing_call="finish")
638    provider = _make_provider()
639    _install_unidentified_track(provider)
640    session_id = "session-finish"
641    assert await provider._start_analysis(session_id, _make_streamdetails(), _make_audio_format())
642    provider._data[session_id].pcm_seconds_fed = 30.0
643
644    assert await provider._finalize(session_id) is None
645
646
647@pytest.mark.asyncio
648async def test_feed_type_error_is_caught_without_a_native_fingerprinter() -> None:
649    """The feed() guard still catches TypeError when no chromaprint errors are registered."""
650    provider = _make_provider()
651    assert provider._fingerprint_errors == ()
652
653    class _RejectingFingerprinter:
654        def feed(self, data: bytes) -> None:
655            raise TypeError("unsupported buffer")
656
657    provider._data["session"] = _AcoustidSessionData(
658        fingerprinter=_RejectingFingerprinter(),
659        sample_rate=44100,
660        channels=2,
661        sample_width=2,
662        track_duration=180,
663    )
664
665    await provider.process_pcm_chunk("session", b"\x00\x01" * 100)
666
667    assert provider._data["session"].error is not None
668
669
670# ---------------------------------------------------------------------------
671# post_analysis side effects
672# ---------------------------------------------------------------------------
673
674
675@pytest.mark.asyncio
676@pytest.mark.parametrize(
677    ("write_tags_back", "expect_tag_writes"),
678    [(False, False), (True, True)],
679    ids=["tags_disabled", "tags_enabled"],
680)
681async def test_post_analysis_persists_and_writes_tags(
682    monkeypatch: pytest.MonkeyPatch,
683    *,
684    write_tags_back: bool,
685    expect_tag_writes: bool,
686) -> None:
687    """post_analysis must always persist to library, and write tags only when enabled."""
688    write_tags = AsyncMock(return_value=True)
689    monkeypatch.setattr(
690        "music_assistant.providers.acoustid_lookup.provider.write_identifier_tags",
691        write_tags,
692    )
693
694    provider = _make_provider(write_tags_back=write_tags_back)
695    streamdetails = _make_streamdetails()
696    analysis = AudioAnalysisData(extra_data={"mbid": "mbid-x", "acoustid": "acoustid-x"})
697
698    consensus_calls: list[Any] = []
699
700    async def fake_consensus(_self: AcoustidLookupProvider, sd: Any, **_kwargs: Any) -> None:
701        consensus_calls.append(sd)
702
703    monkeypatch.setattr(AcoustidLookupProvider, "_maybe_set_album_release_group", fake_consensus)
704
705    await provider.post_analysis(streamdetails, analysis)
706
707    cast("AsyncMock", provider.mass.music.tracks.set_identifiers).assert_awaited_once_with(
708        item_id=streamdetails.item_id,
709        provider_instance_id_or_domain=streamdetails.provider,
710        mbid="mbid-x",
711        acoustid="acoustid-x",
712        isrcs=[],
713    )
714    # Album consensus is a pure DB write and must run regardless of write_tags_back.
715    assert consensus_calls == [streamdetails]
716    if expect_tag_writes:
717        write_tags.assert_awaited_once_with(
718            "/music/track.flac",
719            mbid="mbid-x",
720            acoustid="acoustid-x",
721            isrcs=[],
722            artist_mbids=[],
723        )
724    else:
725        write_tags.assert_not_awaited()
726
727
728@pytest.mark.asyncio
729@pytest.mark.parametrize(
730    ("scenario", "expected_isrcs", "expected_artist_mbids"),
731    [
732        pytest.param(
733            "returns",
734            ["USAB12345678", "USCD87654321"],
735            ["artist-mbid-1", "artist-mbid-2"],
736            id="mb_returns_extras",
737        ),
738        pytest.param("missing", [], [], id="mb_provider_missing"),
739        pytest.param("raises", [], [], id="mb_raises"),
740    ],
741)
742async def test_post_analysis_mb_enrichment(
743    monkeypatch: pytest.MonkeyPatch,
744    *,
745    scenario: str,
746    expected_isrcs: list[str],
747    expected_artist_mbids: list[str],
748) -> None:
749    """ISRCs and artist MBIDs from MusicBrainz get applied as DB/file tags when available."""
750    write_tags = AsyncMock(return_value=True)
751    monkeypatch.setattr(
752        "music_assistant.providers.acoustid_lookup.provider.write_identifier_tags",
753        write_tags,
754    )
755
756    provider = _make_provider(write_tags_back=True)
757    source_mock = MagicMock(write_access=True)
758    mb_provider: MagicMock | None
759    if scenario == "returns":
760        artist_credit = [MagicMock(artist=MagicMock(id=mbid)) for mbid in expected_artist_mbids]
761        mb_recording = MagicMock(isrcs=expected_isrcs, artist_credit=artist_credit)
762        mb_provider = MagicMock()
763        mb_provider.get_recording_details = AsyncMock(return_value=mb_recording)
764    elif scenario == "missing":
765        mb_provider = None
766    else:
767        mb_provider = MagicMock()
768        mb_provider.get_recording_details = AsyncMock(side_effect=aiohttp.ClientError("synthetic"))
769
770    def _get_provider(provider_id: Any, **_kwargs: Any) -> Any:
771        if provider_id == "filesystem_local_test":
772            return source_mock
773        return mb_provider
774
775    provider.mass.get_provider = MagicMock(side_effect=_get_provider)  # type: ignore[method-assign]
776
777    await provider.post_analysis(
778        _make_streamdetails(),
779        AudioAnalysisData(extra_data={"mbid": "mbid-x", "acoustid": "acoustid-x"}),
780    )
781
782    set_ids = cast("AsyncMock", provider.mass.music.tracks.set_identifiers)
783    assert set_ids.await_args is not None
784    assert set_ids.await_args.kwargs["isrcs"] == expected_isrcs
785    write_tags.assert_awaited_once_with(
786        "/music/track.flac",
787        mbid="mbid-x",
788        acoustid="acoustid-x",
789        isrcs=expected_isrcs,
790        artist_mbids=expected_artist_mbids,
791    )
792
793
794@pytest.mark.asyncio
795async def test_post_analysis_consensus_silent_failure(monkeypatch: pytest.MonkeyPatch) -> None:
796    """A crash in the consensus helper must not lose the per-track persistence."""
797    provider = _make_provider()
798
799    async def boom(_self: AcoustidLookupProvider, _streamdetails: Any, **_kwargs: Any) -> None:
800        raise RuntimeError("synthetic")
801
802    monkeypatch.setattr(AcoustidLookupProvider, "_maybe_set_album_release_group", boom)
803
804    await provider.post_analysis(
805        _make_streamdetails(),
806        AudioAnalysisData(
807            extra_data={"mbid": "mbid-x", "acoustid": "acoustid-x", "release_groups": []}
808        ),
809    )
810
811    cast("AsyncMock", provider.mass.music.tracks.set_identifiers).assert_awaited_once()
812
813
814# ---------------------------------------------------------------------------
815# Album-level release-group consensus
816# ---------------------------------------------------------------------------
817
818
819_NEWS_OF_THE_WORLD_QUEEN_RGS = [
820    {
821        "id": "rg-future-boy",
822        "title": "News of the World",
823        "primary_type": "Album",
824        "secondary_types": [],
825        "artists": ["Future Boy"],
826    },
827    {
828        "id": "rg-queen",
829        "title": "News of the World",
830        "primary_type": "Album",
831        "secondary_types": [],
832        "artists": ["Queen"],
833    },
834]
835
836
837@pytest.mark.asyncio
838@pytest.mark.parametrize(
839    (
840        "album_name",
841        "track_provider_instances",
842        "extras",
843        "track_artist",
844        "expected_rg_id",
845    ),
846    [
847        # Strong coverage: every analysed track votes for the same RG.
848        pytest.param(
849            "Silver Thunderbird",
850            ["filesystem_local_test"] * 10,
851            [{"release_groups": [_rg("rg-st")]}] * 10,
852            None,
853            "rg-st",
854            id="full_coverage",
855        ),
856        # Singleton voter with a name-matching RG.
857        pytest.param(
858            "Silver Thunderbird",
859            ["filesystem_local_test"],
860            [{"release_groups": [_rg("rg-st")]}],
861            None,
862            "rg-st",
863            id="singleton_with_match",
864        ),
865        # 5 of 6 voters agree; one outlier on a different RG.
866        pytest.param(
867            "Silver Thunderbird",
868            ["filesystem_local_test"] * 6,
869            [{"release_groups": [_rg("rg-st")]}] * 5
870            + [{"release_groups": [_rg("rg-other", title="Other")]}],
871            None,
872            "rg-st",
873            id="tolerates_one_missing",
874        ),
875        # Quorum denominator excludes tracks served by a different provider.
876        pytest.param(
877            "Silver Thunderbird",
878            ["filesystem_local_test"] * 2 + ["some_other_provider"] * 2,
879            [{"release_groups": [_rg("rg-st")]}] * 2,
880            None,
881            "rg-st",
882            id="quorum_scoped_to_this_provider",
883        ),
884        # Asymmetric substring: user "The Platinum Collection" vs MB's longer form.
885        pytest.param(
886            "The Platinum Collection",
887            ["filesystem_local_test"] * 2,
888            [
889                {
890                    "release_groups": [
891                        _rg(
892                            "rg-platinum",
893                            title="Greatest Hits I, II & III: The Platinum Collection",
894                        )
895                    ]
896                }
897            ]
898            * 2,
899            None,
900            "rg-platinum",
901            id="asymmetric_substring_match",
902        ),
903        # Exact title beats substring when both survive.
904        pytest.param(
905            "The Platinum Collection",
906            ["filesystem_local_test"] * 2,
907            [
908                {
909                    "release_groups": [
910                        _rg(
911                            "rg-substring",
912                            title="Greatest Hits I, II & III: The Platinum Collection",
913                        ),
914                        _rg("rg-exact", title="The Platinum Collection"),
915                    ]
916                }
917            ]
918            * 2,
919            None,
920            "rg-exact",
921            id="exact_beats_substring",
922        ),
923        # Same-titled RGs from different artists; expected_artist picks the matching one.
924        pytest.param(
925            "News Of The World",
926            ["filesystem_local_test"],
927            [{"release_groups": list(_NEWS_OF_THE_WORLD_QUEEN_RGS)}],
928            "Queen",
929            "rg-queen",
930            id="artist_filter_picks_matching_artist",
931        ),
932        # RG with no captured artist info is treated as compatible — older payload shape.
933        pytest.param(
934            "News Of The World",
935            ["filesystem_local_test"],
936            [
937                {
938                    "release_groups": [
939                        {
940                            "id": "rg-unknown-artist",
941                            "title": "News of the World",
942                            "primary_type": "Album",
943                            "secondary_types": [],
944                        }
945                    ]
946                }
947            ],
948            "Queen",
949            "rg-unknown-artist",
950            id="artist_filter_lets_unknown_artist_pass",
951        ),
952    ],
953)
954async def test_consensus_writes_release_group(
955    *,
956    album_name: str,
957    track_provider_instances: list[str],
958    extras: list[dict[str, Any]],
959    track_artist: str | None,
960    expected_rg_id: str,
961) -> None:
962    """Each consensus path that produces a winner writes the RG to the album row."""
963    provider = _make_provider()
964    album = _make_library_album(name=album_name)
965    album_tracks = [
966        _make_album_tracks(1, provider_instance=inst)[0] for inst in track_provider_instances
967    ]
968    _wire_album_for_consensus(provider, album=album, album_tracks=album_tracks, extras=extras)
969    if track_artist is not None:
970        _wire_track_for_mb_lookup(provider, track_name="any", artist_name=track_artist)
971
972    await provider._maybe_set_album_release_group(_make_streamdetails())
973
974    cast("AsyncMock", provider.mass.music.albums.set_release_group).assert_awaited_once_with(
975        42, expected_rg_id
976    )
977
978
979@pytest.mark.asyncio
980@pytest.mark.parametrize(
981    ("album_name", "album_kwargs", "track_provider_instances", "extras"),
982    [
983        # Album already has a release-group — idempotent skip before tally.
984        pytest.param(
985            "Silver Thunderbird",
986            {"existing_rg": "rg-pre-existing"},
987            ["filesystem_local_test"] * 6,
988            [{"release_groups": [_rg("rg-st")]}] * 6,
989            id="album_already_has_releasegroup",
990        ),
991        # Below 50% quorum — 4 of 12 voted.
992        pytest.param(
993            "Silver Thunderbird",
994            {},
995            ["filesystem_local_test"] * 12,
996            [{"release_groups": [_rg("rg-st")]}] * 4,
997            id="below_50pct_quorum",
998        ),
999        # Top coverage below required — 8 voters split 2/2/2/2 across distinct RGs.
1000        pytest.param(
1001            "Silver Thunderbird",
1002            {},
1003            ["filesystem_local_test"] * 8,
1004            [{"release_groups": [_rg(f"rg-{i // 2}", title=f"Title {i // 2}")]} for i in range(8)],
1005            id="top_coverage_below_required",
1006        ),
1007        # No survivor title matches the album — voting tracks share RGs, all wrong names.
1008        pytest.param(
1009            "Silver Thunderbird",
1010            {},
1011            ["filesystem_local_test"] * 6,
1012            [
1013                {
1014                    "release_groups": [
1015                        _rg("rg-here-now", title="Here & Now"),
1016                        _rg("rg-box", title="Box Set"),
1017                    ]
1018                }
1019            ]
1020            * 6,
1021            id="no_survivor_title_matches",
1022        ),
1023        # User tag is more specific than MB title — asymmetric substring rejects.
1024        pytest.param(
1025            "Greatest Hits: 40 Trips Around The Sun",
1026            {},
1027            ["filesystem_local_test"],
1028            [{"release_groups": [_rg("rg-generic", title="Greatest Hits")]}],
1029            id="user_longer_substring_rejected",
1030        ),
1031        # Album has zero library tracks served by this provider.
1032        pytest.param(
1033            "Silver Thunderbird",
1034            {},
1035            ["some_other_provider"] * 3,
1036            [],
1037            id="no_tracks_served_by_this_provider",
1038        ),
1039    ],
1040)
1041async def test_consensus_abstains(
1042    *,
1043    album_name: str,
1044    album_kwargs: dict[str, Any],
1045    track_provider_instances: list[str],
1046    extras: list[dict[str, Any]],
1047) -> None:
1048    """Each guard refuses to write the release-group for the right reason."""
1049    provider = _make_provider()
1050    album = _make_library_album(name=album_name, **album_kwargs)
1051    album_tracks = [
1052        _make_album_tracks(1, provider_instance=inst)[0] for inst in track_provider_instances
1053    ]
1054    _wire_album_for_consensus(provider, album=album, album_tracks=album_tracks, extras=extras)
1055
1056    await provider._maybe_set_album_release_group(_make_streamdetails())
1057
1058    cast("AsyncMock", provider.mass.music.albums.set_release_group).assert_not_awaited()
1059
1060
1061def _wire_track_for_mb_lookup(
1062    provider: AcoustidLookupProvider, *, track_name: str, artist_name: str
1063) -> None:
1064    """Give the library track stub a name and a single named artist."""
1065    library_track = MagicMock()
1066    library_track.name = track_name
1067    library_track.album = MagicMock(item_id="42")
1068    library_track.artists = [MagicMock(name=artist_name)]
1069    library_track.artists[0].name = artist_name
1070    cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
1071        return_value=library_track
1072    )
1073
1074
1075def _install_mb_search(
1076    provider: AcoustidLookupProvider,
1077    *,
1078    rg_id: str | None,
1079    rg_title: str | None,
1080) -> MagicMock:
1081    """Wire mass.get_provider('musicbrainz', ...) to a mock with a fake search()."""
1082    mb_provider = MagicMock()
1083    if rg_id is None:
1084        mb_provider.search = AsyncMock(return_value=None)
1085    else:
1086        rg = MagicMock(id=rg_id, title=rg_title)
1087        artist = MagicMock()
1088        recording = MagicMock()
1089        mb_provider.search = AsyncMock(return_value=(artist, rg, recording))
1090    cast("MagicMock", provider.mass).get_provider = MagicMock(return_value=mb_provider)
1091    return mb_provider
1092
1093
1094@pytest.mark.asyncio
1095@pytest.mark.parametrize(
1096    (
1097        "album_name",
1098        "track_name",
1099        "artist_name",
1100        "extras",
1101        "mb_result",
1102        "expected_mb_kwargs",
1103        "expected_rg",
1104    ),
1105    [
1106        # Consensus has nothing to vote on; MB.search supplies the RG.
1107        pytest.param(
1108            "Alive and Kicking",
1109            "Alive and Kicking",
1110            "Simple Minds",
1111            [{"release_groups": [_rg("rg-unrelated", title="Some Compilation")]}],
1112            ("rg-mb-fallback", "Alive and Kicking"),
1113            {
1114                "artistname": "Simple Minds",
1115                "albumname": "Alive and Kicking",
1116                "trackname": "Alive and Kicking",
1117            },
1118            "rg-mb-fallback",
1119            id="consensus_abstains_mb_writes_rg",
1120        ),
1121        # Hyphen / colon / parens in the album name are flattened before the
1122        # MB query so Lucene's phrase match lines up with MB's stored variant.
1123        pytest.param(
1124            "My Love - Ultimate Essential Collection",
1125            "Beauty and the Beast",
1126            "Céline Dion",
1127            [{"release_groups": [_rg("rg-unrelated", title="Some Compilation")]}],
1128            ("rg-mylove", "My Love: Ultimate Essential Collection"),
1129            {
1130                "artistname": "Céline Dion",
1131                "albumname": "My Love Ultimate Essential Collection",
1132                "trackname": "Beauty and the Beast",
1133            },
1134            "rg-mylove",
1135            id="separators_flattened_for_lucene",
1136        ),
1137        # Consensus has already supplied the right RG; MB.search must not run.
1138        pytest.param(
1139            "Silver Thunderbird",
1140            "Whatever",
1141            "Mary Chapin Carpenter",
1142            [{"release_groups": [_rg("rg-st")]}, {"release_groups": [_rg("rg-st")]}],
1143            ("rg-decoy", "Decoy"),
1144            None,
1145            "rg-st",
1146            id="consensus_succeeds_mb_skipped",
1147        ),
1148        # MB.search returns an RG whose title doesn't match; refuse the write.
1149        pytest.param(
1150            "Alive and Kicking",
1151            "Alive and Kicking",
1152            "Simple Minds",
1153            [{"release_groups": [_rg("rg-unrelated", title="Some Compilation")]}],
1154            ("rg-wrong", "Something Entirely Different"),
1155            {
1156                "artistname": "Simple Minds",
1157                "albumname": "Alive and Kicking",
1158                "trackname": "Alive and Kicking",
1159            },
1160            None,
1161            id="mb_returns_wrong_title_refused",
1162        ),
1163        # No artist on the library track; MB query is skipped cleanly.
1164        pytest.param(
1165            "Alive and Kicking",
1166            "Alive and Kicking",
1167            None,
1168            [{"release_groups": [_rg("rg-unrelated", title="Some Compilation")]}],
1169            ("rg-decoy", "Decoy"),
1170            None,
1171            None,
1172            id="no_artist_mb_skipped",
1173        ),
1174    ],
1175)
1176async def test_mb_fallback(
1177    *,
1178    album_name: str,
1179    track_name: str,
1180    artist_name: str | None,
1181    extras: list[dict[str, Any]],
1182    mb_result: tuple[str, str],
1183    expected_mb_kwargs: dict[str, str] | None,
1184    expected_rg: str | None,
1185) -> None:
1186    """MB.search runs only when consensus abstains and only writes a title-matching RG."""
1187    provider = _make_provider()
1188    album = _make_library_album(name=album_name)
1189    album_tracks = _make_album_tracks(len(extras))
1190    _wire_album_for_consensus(provider, album=album, album_tracks=album_tracks, extras=extras)
1191    if artist_name is not None:
1192        _wire_track_for_mb_lookup(provider, track_name=track_name, artist_name=artist_name)
1193    else:
1194        # Library track stub with empty artists — drives the "no artist" skip path.
1195        library_track = MagicMock()
1196        library_track.name = track_name
1197        library_track.album = MagicMock(item_id="42")
1198        library_track.artists = []
1199        cast("MagicMock", provider.mass.music.tracks).get_library_item_by_prov_id = AsyncMock(
1200            return_value=library_track
1201        )
1202    mb = _install_mb_search(provider, rg_id=mb_result[0], rg_title=mb_result[1])
1203
1204    await provider._maybe_set_album_release_group(_make_streamdetails())
1205
1206    if expected_mb_kwargs is None:
1207        cast("AsyncMock", mb.search).assert_not_awaited()
1208    else:
1209        cast("AsyncMock", mb.search).assert_awaited_once_with(**expected_mb_kwargs)
1210    rg_writer = cast("AsyncMock", provider.mass.music.albums.set_release_group)
1211    if expected_rg is None:
1212        rg_writer.assert_not_awaited()
1213    else:
1214        rg_writer.assert_awaited_once_with(42, expected_rg)
1215
1216
1217# ---------------------------------------------------------------------------
1218# Parser
1219# ---------------------------------------------------------------------------
1220
1221
1222def test_parse_response_aggregates_release_groups() -> None:
1223    """release_groups is the deduped union across recordings, filtered by min_score."""
1224    payload = {
1225        "status": "ok",
1226        "results": [
1227            {
1228                "id": "acoustid-high",
1229                "score": 0.9,
1230                "recordings": [
1231                    {
1232                        "id": "rec-a",
1233                        "title": "Song",
1234                        "releasegroups": [
1235                            {"id": "rg-shared", "title": "Album", "type": "Album"},
1236                            {"id": "rg-a-only", "title": "Sampler", "type": "Compilation"},
1237                            # duplicate of rg-shared must be deduped
1238                            {"id": "rg-shared", "title": "Album", "type": "Album"},
1239                        ],
1240                    },
1241                    {
1242                        "id": "rec-b",
1243                        "releasegroups": [
1244                            {"id": "rg-shared", "title": "Album", "type": "Album"},
1245                            {"id": "rg-b-only", "title": "Reissue", "type": "Album"},
1246                        ],
1247                    },
1248                ],
1249            },
1250            {
1251                "id": "acoustid-low",
1252                "score": 0.3,
1253                "recordings": [
1254                    {
1255                        "id": "rec-low",
1256                        "title": "Song",
1257                        "releasegroups": [
1258                            {"id": "rg-low-score", "title": "Other", "type": "Album"}
1259                        ],
1260                    }
1261                ],
1262            },
1263        ],
1264    }
1265    _score, _acoustid, _mbid, _candidates, _matched, release_groups = _parse_response(
1266        payload, min_score=0.5
1267    )
1268    assert {rg["id"] for rg in release_groups} == {"rg-shared", "rg-a-only", "rg-b-only"}
1269
1270
1271@pytest.mark.parametrize(
1272    (
1273        "expected_track_title",
1274        "expected_album_title",
1275        "payload",
1276        "expected_mbid",
1277        "expected_acoustid",
1278    ),
1279    [
1280        # No track title supplied — filter is a no-op, score picks the winner.
1281        pytest.param(
1282            None,
1283            None,
1284            {
1285                "status": "ok",
1286                "results": [
1287                    {
1288                        "id": "acoustid-x",
1289                        "score": 0.9,
1290                        "recordings": [
1291                            {"id": "rec-x", "title": "Anything", "releases": [{"id": "r"}]}
1292                        ],
1293                    }
1294                ],
1295            },
1296            "rec-x",
1297            "acoustid-x",
1298            id="no_hint_falls_back_to_score",
1299        ),
1300        # Hint matches one recording — that recording wins.
1301        pytest.param(
1302            "Ventura Highway",
1303            None,
1304            {
1305                "status": "ok",
1306                "results": [
1307                    {
1308                        "id": "acoustid-x",
1309                        "score": 0.95,
1310                        "recordings": [
1311                            {"id": "rec-match", "title": "Ventura Highway"},
1312                            {"id": "rec-other", "title": "Other"},
1313                        ],
1314                    }
1315                ],
1316            },
1317            "rec-match",
1318            "acoustid-x",
1319            id="match_picks_correct_recording",
1320        ),
1321        # No recording matches the hint anywhere — refuse the match.
1322        pytest.param(
1323            "Ventura Highway",
1324            None,
1325            {
1326                "status": "ok",
1327                "results": [
1328                    {
1329                        "id": "acoustid-x",
1330                        "score": 0.97,
1331                        "recordings": [{"id": "rec-wrong", "title": "Trouble"}],
1332                    }
1333                ],
1334            },
1335            None,
1336            None,
1337            id="no_match_refuses",
1338        ),
1339        # Lower-score result whose recording title matches must beat higher-score non-match.
1340        pytest.param(
1341            "Ventura Highway",
1342            None,
1343            {
1344                "status": "ok",
1345                "results": [
1346                    {
1347                        "id": "acoustid-high",
1348                        "score": 0.99,
1349                        "recordings": [{"id": "rec-wrong", "title": "Trouble"}],
1350                    },
1351                    {
1352                        "id": "acoustid-low",
1353                        "score": 0.85,
1354                        "recordings": [{"id": "rec-right", "title": "Ventura Highway"}],
1355                    },
1356                ],
1357            },
1358            "rec-right",
1359            "acoustid-low",
1360            id="prefers_title_match_over_score",
1361        ),
1362        # MB has the canonical title; user has a medley/prefix that contains it.
1363        pytest.param(
1364            "Water Song / Janie's Got a Gun",
1365            None,
1366            {
1367                "status": "ok",
1368                "results": [
1369                    {
1370                        "id": "acoustid-x",
1371                        "score": 0.95,
1372                        "recordings": [{"id": "rec-jgg", "title": "Janie's Got a Gun"}],
1373                    }
1374                ],
1375            },
1376            "rec-jgg",
1377            "acoustid-x",
1378            id="substring_fallback_matches_subtitle",
1379        ),
1380        # Single-word title must not match a longer phrase containing it (trivial-collision guard).
1381        pytest.param(
1382            "Trouble",
1383            None,
1384            {
1385                "status": "ok",
1386                "results": [
1387                    {
1388                        "id": "acoustid-x",
1389                        "score": 0.97,
1390                        "recordings": [{"id": "rec-im-in-trouble", "title": "I'm In Trouble"}],
1391                    }
1392                ],
1393            },
1394            None,
1395            None,
1396            id="substring_fallback_rejects_single_word",
1397        ),
1398        # Two results both title-match the track; the album hint picks the right release.
1399        pytest.param(
1400            "Ventura Highway",
1401            "History: America's Greatest Hits",
1402            {
1403                "status": "ok",
1404                "results": [
1405                    {
1406                        "id": "acoustid-5.1mix",
1407                        "score": 0.99,
1408                        "recordings": [
1409                            {
1410                                "id": "rec-homecoming",
1411                                "title": "Ventura Highway",
1412                                "releases": [{"id": "rel-h", "title": "Homecoming"}],
1413                            }
1414                        ],
1415                    },
1416                    {
1417                        "id": "acoustid-history",
1418                        "score": 0.85,
1419                        "recordings": [
1420                            {
1421                                "id": "rec-history",
1422                                "title": "Ventura Highway",
1423                                "releases": [
1424                                    {"id": "rel-hist", "title": "History: America's Greatest Hits"}
1425                                ],
1426                            }
1427                        ],
1428                    },
1429                ],
1430            },
1431            "rec-history",
1432            "acoustid-history",
1433            id="prefers_album_match_when_track_matches_tie",
1434        ),
1435        # Remaster suffix stripping — user tag and MB title differ only by a
1436        # version qualifier; pre-normalise strip should make them match.
1437        *(
1438            pytest.param(
1439                user_title,
1440                None,
1441                {
1442                    "status": "ok",
1443                    "results": [
1444                        {
1445                            "id": "acoustid-x",
1446                            "score": 0.99,
1447                            "recordings": [{"id": "rec-africa", "title": mb_title}],
1448                        }
1449                    ],
1450                },
1451                "rec-africa",
1452                "acoustid-x",
1453                id=f"remaster_strip_{case_id}",
1454            )
1455            for case_id, user_title, mb_title in (
1456                ("user_year_paren", "Africa (2016 Remaster)", "Africa"),
1457                ("mb_year_paren", "Africa", "Africa (2018 Remaster)"),
1458                ("user_remastered_year", "Africa (Remastered 2011)", "Africa"),
1459                ("user_hyphen", "Africa - Remastered", "Africa"),
1460                ("user_hyphen_year", "Africa - 2018 Remaster", "Africa"),
1461                ("both_variants", "Africa (Remaster)", "Africa (2018 Remaster)"),
1462                ("ampersand_user_word", "Alive And Kicking", "Alive & Kicking"),
1463                ("ampersand_user_amp", "Alive & Kicking", "Alive And Kicking"),
1464            )
1465        ),
1466        # Tiered album match: when one recording's release exactly matches the
1467        # user's specific album and another only substring-matches a generic
1468        # release, the exact match wins despite a lower fingerprint score.
1469        pytest.param(
1470            "Africa",
1471            "Greatest Hits: 40 Trips Around The Sun",
1472            {
1473                "status": "ok",
1474                "results": [
1475                    {
1476                        "id": "acoustid-substring",
1477                        "score": 0.99,
1478                        "recordings": [
1479                            {
1480                                "id": "rec-substring",
1481                                "title": "Africa",
1482                                "releases": [{"id": "rel-gh", "title": "Greatest Hits"}],
1483                            }
1484                        ],
1485                    },
1486                    {
1487                        "id": "acoustid-exact",
1488                        "score": 0.85,
1489                        "recordings": [
1490                            {
1491                                "id": "rec-exact",
1492                                "title": "Africa",
1493                                "releases": [
1494                                    {
1495                                        "id": "rel-40trips",
1496                                        "title": "Greatest Hits: 40 Trips Around The Sun",
1497                                    }
1498                                ],
1499                            }
1500                        ],
1501                    },
1502                ],
1503            },
1504            "rec-exact",
1505            "acoustid-exact",
1506            id="exact_album_match_beats_substring",
1507        ),
1508    ],
1509)
1510def test_parse_response_track_name_behaviour(
1511    *,
1512    expected_track_title: str | None,
1513    expected_album_title: str | None,
1514    payload: dict[str, Any],
1515    expected_mbid: str | None,
1516    expected_acoustid: str | None,
1517) -> None:
1518    """expected_track_title is a hard filter; expected_album_title disambiguates ties."""
1519    _score, acoustid, mbid, _candidates, _matched, _rgs = _parse_response(
1520        payload,
1521        expected_track_title=expected_track_title,
1522        expected_album_title=expected_album_title,
1523    )
1524    assert mbid == expected_mbid
1525    assert acoustid == expected_acoustid
1526