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