music-assistant-server

1"""Tests for AlbumsController provider matching (explicit, IO-capable enrichment)."""
2
3from __future__ import annotations
4
5import logging
6from contextlib import contextmanager
7from dataclasses import dataclass
8from types import SimpleNamespace
9from typing import TYPE_CHECKING
10from unittest.mock import AsyncMock, Mock, patch
11
12from music_assistant_models.enums import ExternalID
13from music_assistant_models.errors import MediaNotFoundError, RetriesExhausted
14from music_assistant_models.media_items import (
15    Album,
16    Artist,
17    ProviderMapping,
18    Track,
19    UniqueList,
20)
21
22from music_assistant.controllers.music.media.albums import AlbumsController
23from music_assistant.helpers.compare import AlbumMatchEvidence
24
25if TYPE_CHECKING:
26    from collections.abc import Iterator, Sequence
27
28MB_ALBUM_ID = "11111111-1111-1111-1111-111111111111"
29BASE_BARCODE = "888072439412"
30OTHER_BARCODE = "075678643224"
31THIRD_BARCODE = "093624912514"
32# the base library album is linked to one existing (already-loaded) provider
33BASE_MAPPING = ProviderMapping(
34    item_id="base-prov", provider_domain="tidal", provider_instance="tidal_1"
35)
36
37
38# ---------------------------------------------------------------------------
39# builders
40# ---------------------------------------------------------------------------
41
42
43def _artist() -> Artist:
44    """Return the single album artist shared by every fixture album."""
45    return Artist(
46        item_id="artist",
47        provider="library",
48        name="Sigur Rós",
49        provider_mappings={
50            ProviderMapping(item_id="artist", provider_domain="test", provider_instance="library")
51        },
52    )
53
54
55def _album(
56    item_id: str,
57    provider: str,
58    *,
59    name: str = "( )",
60    version: str = "",
61    year: int | None = 2022,
62    external_ids: set[tuple[ExternalID, str]] | None = None,
63    barcodes: Sequence[str] = (),
64    mappings: Sequence[ProviderMapping] | None = None,
65) -> Album:
66    """Build an Album for match-provider tests."""
67    ext = set(external_ids or set())
68    ext.update((ExternalID.BARCODE, barcode) for barcode in barcodes)
69    if mappings is None:
70        mappings = [
71            ProviderMapping(item_id=item_id, provider_domain=provider, provider_instance=provider)
72        ]
73    return Album(
74        item_id=item_id,
75        provider=provider,
76        name=name,
77        version=version,
78        year=year,
79        external_ids=ext,
80        artists=UniqueList([_artist()]),
81        provider_mappings=set(mappings),
82    )
83
84
85def _library_album(
86    *,
87    version: str = "",
88    external_ids: set[tuple[ExternalID, str]] | None = None,
89    barcodes: Sequence[str] = (),
90    mappings: Sequence[ProviderMapping] = (BASE_MAPPING,),
91) -> Album:
92    """Build the base (library) album under match."""
93    return _album(
94        "lib1",
95        "library",
96        version=version,
97        external_ids=external_ids,
98        barcodes=barcodes,
99        mappings=mappings,
100    )
101
102
103def _track(
104    number: int, *, isrc_prefix: str = "USRC17607", duration: int = 200, disc_number: int = 1
105) -> Track:
106    """Build one ordered album track with a distinct ISRC per position."""
107    return Track(
108        item_id=str(number),
109        provider="spotify_1",
110        name=f"Track {number}",
111        duration=duration,
112        disc_number=disc_number,
113        track_number=number,
114        external_ids={(ExternalID.ISRC, f"{isrc_prefix}{number:03d}")},
115        provider_mappings={
116            ProviderMapping(
117                item_id=str(number), provider_domain="spotify", provider_instance="spotify_1"
118            )
119        },
120    )
121
122
123def _tracklist(count: int, *, isrc_prefix: str = "USRC17607", disc_number: int = 1) -> list[Track]:
124    """Build an ordered tracklist of the given length."""
125    return [
126        _track(number, isrc_prefix=isrc_prefix, disc_number=disc_number)
127        for number in range(1, count + 1)
128    ]
129
130
131def _mb_release(release_id: str, release_group_id: str) -> SimpleNamespace:
132    """Return a stand-in for a parsed MusicBrainz release."""
133    return SimpleNamespace(id=release_id, release_group=SimpleNamespace(id=release_group_id))
134
135
136def _provider() -> Mock:
137    """Return a mock streaming MusicProvider for matching."""
138    provider = Mock()
139    provider.name = "Spotify"
140    provider.instance_id = "spotify_1"
141    provider.domain = "spotify"
142    return provider
143
144
145def _mb(**releases_by_barcode: list[SimpleNamespace]) -> Mock:
146    """Return a mock MusicBrainz provider answering barcode lookups from a mapping."""
147    lookup = {BASE_BARCODE: [], OTHER_BARCODE: [], THIRD_BARCODE: [], **releases_by_barcode}
148    musicbrainz = Mock()
149    musicbrainz.get_releases_by_barcode = AsyncMock(side_effect=lambda barcode: lookup[barcode])
150    return musicbrainz
151
152
153def _loaded_provider(instance_id: str, *, available: bool = True) -> Mock:
154    """Return a mock provider instance for the loaded-provider registry."""
155    provider = Mock()
156    provider.instance_id = instance_id
157    provider.available = available
158    return provider
159
160
161@dataclass
162class _Harness:
163    """A controller under test together with its mocked IO boundaries."""
164
165    ctrl: AlbumsController
166    search: AsyncMock
167    get_provider_item: AsyncMock
168    get_provider_album_tracks: AsyncMock
169    get_provider: Mock
170    provider: Mock
171
172    async def match(self, db_album: Album, *, strict: bool = True) -> list[ProviderMapping]:
173        """Match against the single (streaming) provider the harness owns."""
174        return await self.ctrl.match_provider(db_album, self.provider, strict)
175
176    def album_track_calls(self) -> list[str]:
177        """Return the album item ids passed to each provider album-track lookup."""
178        return [call.args[0] for call in self.get_provider_album_tracks.await_args_list]
179
180
181@contextmanager
182def _harness(
183    *,
184    search_results: Sequence[Album],
185    provider_items: dict[str, Album],
186    provider_album_tracks: dict[str, list[Track] | Exception] | None = None,
187    musicbrainz: Mock | None = None,
188    loaded_instances: Sequence[str] = ("tidal_1",),
189    provider_registry: dict[str, Mock] | None = None,
190) -> Iterator[_Harness]:
191    """
192    Yield an AlbumsController with every IO boundary mocked.
193
194    :param search_results: Sparse album search results returned to the prefilter.
195    :param provider_items: Full provider albums keyed by their item id.
196    :param provider_album_tracks: Provider album tracks (or an error to raise) keyed by
197        album item id; the base album's mapping ids resolve here too.
198    :param musicbrainz: Optional mock MusicBrainz provider.
199    :param loaded_instances: Provider instance ids considered currently loaded and available.
200    :param provider_registry: Explicit instance-id -> provider mock overrides.
201    """
202    album_tracks = provider_album_tracks or {}
203    registry = {instance: _loaded_provider(instance) for instance in loaded_instances}
204    registry.update(provider_registry or {})
205
206    def _get_provider(domain: str, return_unavailable: bool = False) -> object | None:
207        if domain == "musicbrainz":
208            return musicbrainz
209        provider = registry.get(domain)
210        if provider is None:
211            return None
212        return provider if (return_unavailable or provider.available) else None
213
214    mass = Mock()
215    mass.get_provider = Mock(side_effect=_get_provider)
216    ctrl = AlbumsController.__new__(AlbumsController)
217    ctrl.logger = logging.getLogger("test.albums.match")
218    ctrl.mass = mass
219    search = AsyncMock(return_value=list(search_results))
220    get_provider_item = AsyncMock(
221        side_effect=lambda item_id, _provider, **_kwargs: provider_items[item_id]
222    )
223
224    async def _album_tracks(item_id: str, *_rest: object) -> list[Track]:
225        result = album_tracks.get(item_id, [])
226        if isinstance(result, Exception):
227            raise result
228        return list(result)
229
230    # a single recorder backs both the base mapping lookup (_get_provider_album_tracks)
231    # and the candidate lookup (provider.get_album_tracks), so their calls stay ordered
232    get_provider_album_tracks = AsyncMock(side_effect=_album_tracks)
233    provider = _provider()
234    provider.get_album_tracks = get_provider_album_tracks
235    with patch.multiple(
236        ctrl,
237        search=search,
238        get_provider_item=get_provider_item,
239        _get_provider_album_tracks=get_provider_album_tracks,
240    ):
241        yield _Harness(
242            ctrl, search, get_provider_item, get_provider_album_tracks, mass.get_provider, provider
243        )
244
245
246# ---------------------------------------------------------------------------
247# search-result prefilter
248# ---------------------------------------------------------------------------
249
250
251async def test_clear_no_match_search_result_skips_full_fetch() -> None:
252    """A confidently non-matching search result is dropped before any full fetch."""
253    other = _album("s1", "spotify_1", name="Takk...")
254    with _harness(search_results=[other], provider_items={}) as harness:
255        matches = await harness.match(_library_album())
256
257    assert matches == []
258    harness.get_provider_item.assert_not_awaited()
259
260
261async def test_insufficient_search_result_proceeds_to_one_full_fetch() -> None:
262    """An ambiguous search result is confirmed against exactly one full provider album."""
263    base = _library_album(external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
264    sparse = _album("s1", "spotify_1", version="Remaster")
265    full = _album("s1", "spotify_1", external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
266    with _harness(search_results=[sparse], provider_items={"s1": full}) as harness:
267        matches = await harness.match(base)
268
269    assert [mapping.item_id for mapping in matches] == ["s1"]
270    harness.get_provider_item.assert_awaited_once()
271
272
273async def test_full_item_match_uses_no_track_or_musicbrainz_calls() -> None:
274    """A full album that matches on metadata never fetches tracks or hits MusicBrainz."""
275    musicbrainz = _mb()
276    base = _library_album(external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
277    sparse = _album("s1", "spotify_1", version="Remaster")
278    full = _album("s1", "spotify_1", external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
279    with _harness(
280        search_results=[sparse], provider_items={"s1": full}, musicbrainz=musicbrainz
281    ) as harness:
282        matches = await harness.match(base)
283
284    assert [mapping.item_id for mapping in matches] == ["s1"]
285    harness.get_provider_album_tracks.assert_not_awaited()
286    musicbrainz.get_releases_by_barcode.assert_not_awaited()
287
288
289# ---------------------------------------------------------------------------
290# track-fingerprint resolution
291# ---------------------------------------------------------------------------
292
293
294async def test_ambiguous_albums_resolve_via_track_fingerprints() -> None:
295    """The 14-track remaster vs deluxe-remaster case matches once tracklists agree."""
296    musicbrainz = _mb()
297    base = _library_album(version="2022 Remaster")
298    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
299    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
300    with _harness(
301        search_results=[sparse],
302        provider_items={"s1": full},
303        provider_album_tracks={"base-prov": _tracklist(14), "s1": _tracklist(14)},
304        musicbrainz=musicbrainz,
305    ) as harness:
306        matches = await harness.match(base)
307
308    assert [mapping.item_id for mapping in matches] == ["s1"]
309    # base tracklist comes from the existing provider mapping, candidate from the match target
310    assert harness.album_track_calls() == ["base-prov", "s1"]
311    # a decisive fingerprint match must not fall through to MusicBrainz
312    musicbrainz.get_releases_by_barcode.assert_not_awaited()
313
314
315async def test_different_track_counts_do_not_map() -> None:
316    """A complete 8-track album and a 14-track edition are never merged by fingerprints."""
317    base = _library_album(version="2022 Remaster")
318    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
319    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
320    with _harness(
321        search_results=[sparse],
322        provider_items={"s1": full},
323        provider_album_tracks={"base-prov": _tracklist(8), "s1": _tracklist(14)},
324    ) as harness:
325        assert await harness.match(base) == []
326
327
328async def test_conflicting_isrc_fingerprints_do_not_map() -> None:
329    """Same-length tracklists with conflicting ISRCs are a different recording."""
330    base = _library_album(version="2022 Remaster")
331    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
332    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
333    with _harness(
334        search_results=[sparse],
335        provider_items={"s1": full},
336        provider_album_tracks={
337            "base-prov": _tracklist(14, isrc_prefix="USRC17607"),
338            "s1": _tracklist(14, isrc_prefix="USRC28718"),
339        },
340    ) as harness:
341        assert await harness.match(base) == []
342
343
344async def test_base_tracks_fetched_once_across_candidates() -> None:
345    """The base tracklist is fetched once even when several candidates need it."""
346    base = _library_album(version="2022 Remaster")
347    search_results = [
348        _album(f"s{index}", "spotify_1", version="Deluxe 2022 Remaster") for index in range(1, 4)
349    ]
350    provider_items = {album.item_id: album for album in search_results}
351    with _harness(
352        search_results=search_results,
353        provider_items=provider_items,
354        # every candidate stays ambiguous (no tracks to compare against)
355        provider_album_tracks={"base-prov": _tracklist(14)},
356    ) as harness:
357        await harness.match(base)
358
359    # the base mapping is only fetched once, regardless of the number of candidates
360    assert harness.album_track_calls().count("base-prov") == 1
361
362
363async def test_base_tracks_use_first_trustworthy_loaded_mapping() -> None:
364    """Base resolution deterministically skips unloaded/untrustworthy mappings."""
365    base = _library_album(
366        version="2022 Remaster",
367        mappings=[
368            ProviderMapping(item_id="a", provider_domain="aaa", provider_instance="aaa_1"),
369            ProviderMapping(item_id="b", provider_domain="bbb", provider_instance="bbb_1"),
370            ProviderMapping(item_id="c", provider_domain="ccc", provider_instance="ccc_1"),
371        ],
372    )
373    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
374    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
375    with _harness(
376        search_results=[sparse],
377        provider_items={"s1": full},
378        provider_album_tracks={
379            # "a" is on an unloaded instance and must never be fetched
380            "a": _tracklist(14),
381            # "b" is loaded but its positions are untrustworthy (no disc numbers)
382            "b": _tracklist(14, disc_number=0),
383            "c": _tracklist(14),
384            "s1": _tracklist(14),
385        },
386        loaded_instances=("bbb_1", "ccc_1", "tidal_1"),
387    ) as harness:
388        matches = await harness.match(base)
389
390    assert [mapping.item_id for mapping in matches] == ["s1"]
391    # unloaded "a" skipped before any fetch; "b" fetched but rejected; "c" used
392    assert harness.album_track_calls() == ["b", "c", "s1"]
393
394
395async def test_base_mapping_not_found_is_skipped() -> None:
396    """A base mapping whose tracklist lookup 404s is skipped for the next mapping."""
397    base = _library_album(
398        version="2022 Remaster",
399        mappings=[
400            ProviderMapping(item_id="gone", provider_domain="aaa", provider_instance="aaa_1"),
401            ProviderMapping(item_id="ok", provider_domain="bbb", provider_instance="bbb_1"),
402        ],
403    )
404    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
405    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
406    with _harness(
407        search_results=[sparse],
408        provider_items={"s1": full},
409        provider_album_tracks={
410            "gone": MediaNotFoundError("gone"),
411            "ok": _tracklist(14),
412            "s1": _tracklist(14),
413        },
414        loaded_instances=("aaa_1", "bbb_1"),
415    ) as harness:
416        matches = await harness.match(base)
417
418    assert [mapping.item_id for mapping in matches] == ["s1"]
419    assert harness.album_track_calls() == ["gone", "ok", "s1"]
420
421
422async def test_base_mapping_transient_error_is_skipped() -> None:
423    """A transient base tracklist failure skips to the next mapping."""
424    base = _library_album(
425        version="2022 Remaster",
426        mappings=[
427            ProviderMapping(item_id="flaky", provider_domain="aaa", provider_instance="aaa_1"),
428            ProviderMapping(item_id="ok", provider_domain="bbb", provider_instance="bbb_1"),
429        ],
430    )
431    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
432    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
433    with _harness(
434        search_results=[sparse],
435        provider_items={"s1": full},
436        provider_album_tracks={
437            "flaky": RetriesExhausted("flaky"),
438            "ok": _tracklist(14),
439            "s1": _tracklist(14),
440        },
441        loaded_instances=("aaa_1", "bbb_1"),
442    ) as harness:
443        matches = await harness.match(base)
444
445    assert [mapping.item_id for mapping in matches] == ["s1"]
446    assert harness.album_track_calls() == ["flaky", "ok", "s1"]
447
448
449async def test_base_mapping_skipped_when_exact_instance_not_loaded() -> None:
450    """A mapping whose exact instance is not loaded is skipped, never a same-domain fallback."""
451    base = _library_album(
452        version="2022 Remaster",
453        mappings=[
454            ProviderMapping(item_id="wrong", provider_domain="aaa", provider_instance="aaa_1"),
455            ProviderMapping(item_id="ok", provider_domain="bbb", provider_instance="bbb_1"),
456        ],
457    )
458    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
459    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
460    with _harness(
461        search_results=[sparse],
462        provider_items={"s1": full},
463        provider_album_tracks={"wrong": _tracklist(14), "ok": _tracklist(14), "s1": _tracklist(14)},
464        loaded_instances=("bbb_1",),
465        # aaa_1 resolves (same-domain fallback) to a different instance, so it must be skipped
466        provider_registry={"aaa_1": _loaded_provider("other_1")},
467    ) as harness:
468        matches = await harness.match(base)
469
470    assert [mapping.item_id for mapping in matches] == ["s1"]
471    # "wrong" is never fetched because its exact instance isn't the one that resolved
472    assert harness.album_track_calls() == ["ok", "s1"]
473
474
475async def test_candidate_tracklist_not_found_falls_through_to_musicbrainz() -> None:
476    """A candidate tracklist that 404s is treated as absent so MusicBrainz can decide."""
477    musicbrainz = _mb(
478        **{
479            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
480            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
481        }
482    )
483    base = _library_album(version="2022 Remaster", barcodes=[BASE_BARCODE])
484    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
485    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
486    with _harness(
487        search_results=[sparse],
488        provider_items={"s1": full},
489        provider_album_tracks={"base-prov": _tracklist(14), "s1": MediaNotFoundError("s1")},
490        musicbrainz=musicbrainz,
491    ) as harness:
492        matches = await harness.match(base)
493
494    # MusicBrainz was consulted despite the tracklist failure and its verdict
495    # (disjoint release groups) was applied
496    assert matches == []
497    musicbrainz.get_releases_by_barcode.assert_awaited()
498
499
500async def test_candidate_tracklist_transient_error_falls_through_to_musicbrainz() -> None:
501    """A transient candidate tracklist failure is treated as absent so MusicBrainz can decide."""
502    musicbrainz = _mb(
503        **{
504            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
505            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
506        }
507    )
508    base = _library_album(version="2022 Remaster", barcodes=[BASE_BARCODE])
509    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
510    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
511    with _harness(
512        search_results=[sparse],
513        provider_items={"s1": full},
514        provider_album_tracks={"base-prov": _tracklist(14), "s1": TimeoutError()},
515        musicbrainz=musicbrainz,
516    ) as harness:
517        matches = await harness.match(base)
518
519    # MusicBrainz was consulted despite the tracklist failure and its verdict
520    # (disjoint release groups) was applied
521    assert matches == []
522    musicbrainz.get_releases_by_barcode.assert_awaited()
523
524
525async def test_candidate_tracklist_uses_matched_provider_not_same_domain_fallback() -> None:
526    """The candidate tracklist is fetched from the matched provider, never a same-domain fallback."""
527    # a second instance of the candidate's domain is registered in the loaded-provider
528    # registry; re-resolving the candidate through it would fingerprint against the wrong
529    # account and reject the correct match
530    fallback = _loaded_provider("spotify_other")
531    fallback.get_album_tracks = AsyncMock(return_value=_tracklist(14, isrc_prefix="USRC28718"))
532    base = _library_album(version="2022 Remaster")
533    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
534    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
535    with _harness(
536        search_results=[sparse],
537        provider_items={"s1": full},
538        provider_album_tracks={"base-prov": _tracklist(14)},
539        provider_registry={"spotify_1": fallback},
540    ) as harness:
541        # the matched provider itself returns the correct, agreeing tracklist
542        harness.provider.get_album_tracks = AsyncMock(return_value=_tracklist(14))
543        matches = await harness.match(base)
544
545    assert [mapping.item_id for mapping in matches] == ["s1"]
546    harness.provider.get_album_tracks.assert_awaited_once_with("s1")
547    # the same-domain fallback is never consulted and the candidate domain is never re-resolved
548    fallback.get_album_tracks.assert_not_awaited()
549    assert all(call.args[0] != "spotify_1" for call in harness.get_provider.call_args_list)
550
551
552# ---------------------------------------------------------------------------
553# MusicBrainz last-resort evidence
554# ---------------------------------------------------------------------------
555
556
557@contextmanager
558def _mb_harness(
559    musicbrainz: Mock | None,
560    *,
561    base_barcodes: Sequence[str] = (BASE_BARCODE,),
562    compare_barcodes: Sequence[str] = (OTHER_BARCODE,),
563) -> Iterator[tuple[_Harness, Album]]:
564    """Yield a harness plus base album whose only candidate is ambiguous past fingerprints."""
565    base = _library_album(version="2022 Remaster", barcodes=base_barcodes)
566    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=compare_barcodes)
567    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=compare_barcodes)
568    with _harness(
569        search_results=[sparse],
570        provider_items={"s1": full},
571        # base tracklist present but candidate has none: fingerprint stays inconclusive
572        provider_album_tracks={"base-prov": _tracklist(14), "s1": []},
573        musicbrainz=musicbrainz,
574    ) as harness:
575        yield harness, base
576
577
578async def test_shared_barcode_resolves_ambiguity_without_musicbrainz() -> None:
579    """A shared barcode resolves an ambiguous edition wording without any lookups."""
580    musicbrainz = _mb()
581    with _mb_harness(musicbrainz, compare_barcodes=(BASE_BARCODE,)) as (harness, base):
582        matches = await harness.match(base)
583
584    assert [mapping.item_id for mapping in matches] == ["s1"]
585    musicbrainz.get_releases_by_barcode.assert_not_awaited()
586
587
588async def test_musicbrainz_multiple_releases_per_barcode_abstains() -> None:
589    """A barcode resolving to several releases is ambiguous and must not match."""
590    musicbrainz = _mb(
591        **{
592            BASE_BARCODE: [_mb_release("rel-1", "rg-1"), _mb_release("rel-2", "rg-1")],
593            OTHER_BARCODE: [_mb_release("rel-3", "rg-1")],
594        }
595    )
596    with _mb_harness(musicbrainz) as (harness, base):
597        assert await harness.match(base) == []
598
599
600async def test_musicbrainz_multiple_barcodes_are_all_considered() -> None:
601    """Every canonical barcode on each album is used, not just the first."""
602    musicbrainz = _mb(
603        **{
604            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
605            THIRD_BARCODE: [_mb_release("rel-9", "rg-9")],
606            OTHER_BARCODE: [_mb_release("rel-2", "rg-9")],
607        }
608    )
609    with _mb_harness(
610        musicbrainz, base_barcodes=(BASE_BARCODE, THIRD_BARCODE), compare_barcodes=(OTHER_BARCODE,)
611    ) as (harness, base):
612        assert await harness.match(base) == []
613    queried = {call.args[0] for call in musicbrainz.get_releases_by_barcode.await_args_list}
614    assert queried == {BASE_BARCODE, OTHER_BARCODE, THIRD_BARCODE}
615
616
617async def test_musicbrainz_shared_release_group_alone_does_not_match() -> None:
618    """Different specific releases in one release group must not identify an edition."""
619    musicbrainz = _mb(
620        **{
621            BASE_BARCODE: [_mb_release("rel-1", "rg-9")],
622            OTHER_BARCODE: [_mb_release("rel-2", "rg-9")],
623        }
624    )
625    with _mb_harness(musicbrainz, compare_barcodes=(OTHER_BARCODE,)) as (harness, base):
626        assert await harness.match(base) == []
627
628
629async def test_musicbrainz_disjoint_release_groups_do_not_match() -> None:
630    """Barcodes in entirely different release groups are negative evidence."""
631    musicbrainz = _mb(
632        **{
633            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
634            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
635        }
636    )
637    with _mb_harness(musicbrainz, compare_barcodes=(OTHER_BARCODE,)) as (harness, base):
638        assert await harness.match(base) == []
639
640
641async def test_musicbrainz_unresolved_barcode_abstains() -> None:
642    """A barcode MusicBrainz cannot resolve abstains instead of guessing."""
643    with _mb_harness(_mb()) as (harness, base):
644        assert await harness.match(base) == []
645
646
647async def test_musicbrainz_transport_error_abstains() -> None:
648    """A MusicBrainz outage abstains rather than aborting the whole match."""
649    musicbrainz = Mock()
650    musicbrainz.get_releases_by_barcode = AsyncMock(side_effect=TimeoutError())
651    with _mb_harness(musicbrainz) as (harness, base):
652        assert await harness.match(base) == []
653
654
655async def test_musicbrainz_skipped_without_a_configured_provider() -> None:
656    """With no MusicBrainz provider configured the match simply abstains."""
657    with _mb_harness(None) as (harness, base):
658        assert await harness.match(base) == []
659
660
661def _mb_evidence_ctrl(musicbrainz: Mock) -> AlbumsController:
662    """Return a bare controller wired to a mock MusicBrainz provider."""
663    mass = Mock()
664    mass.get_provider = Mock(
665        side_effect=lambda domain: musicbrainz if domain == "musicbrainz" else None
666    )
667    ctrl = AlbumsController.__new__(AlbumsController)
668    ctrl.logger = logging.getLogger("test.albums.mb")
669    ctrl.mass = mass
670    return ctrl
671
672
673async def test_musicbrainz_evidence_shared_single_release_matches() -> None:
674    """A shared barcode resolving to exactly one specific release is positive evidence."""
675    musicbrainz = _mb(**{BASE_BARCODE: [_mb_release("rel-1", "rg-1")]})
676    ctrl = _mb_evidence_ctrl(musicbrainz)
677    base = _library_album(barcodes=(BASE_BARCODE,))
678    compare = _album("s1", "spotify_1", barcodes=(BASE_BARCODE,))
679
680    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.MATCH
681
682
683async def test_musicbrainz_evidence_ambiguous_barcode_is_not_release_identity() -> None:
684    """A shared barcode resolving to several releases never identifies a specific release."""
685    musicbrainz = _mb(
686        **{BASE_BARCODE: [_mb_release("rel-1", "rg-1"), _mb_release("rel-2", "rg-1")]}
687    )
688    ctrl = _mb_evidence_ctrl(musicbrainz)
689    base = _library_album(barcodes=(BASE_BARCODE,))
690    compare = _album("s1", "spotify_1", barcodes=(BASE_BARCODE,))
691
692    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.INSUFFICIENT
693
694
695async def test_musicbrainz_evidence_all_resolved_disjoint_groups_reject() -> None:
696    """Disjoint release groups are negative evidence when every barcode resolved."""
697    musicbrainz = _mb(
698        **{
699            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
700            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
701        }
702    )
703    ctrl = _mb_evidence_ctrl(musicbrainz)
704    base = _library_album(barcodes=(BASE_BARCODE,))
705    compare = _album("s1", "spotify_1", barcodes=(OTHER_BARCODE,))
706
707    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.NO_MATCH
708
709
710async def test_musicbrainz_evidence_unresolved_barcode_blocks_disjoint_rejection() -> None:
711    """An unresolved barcode leaves the group sets incomplete, so it cannot reject."""
712    musicbrainz = _mb(
713        **{
714            # BASE_BARCODE stays unresolved ([]); THIRD_BARCODE resolves to a distinct group
715            THIRD_BARCODE: [_mb_release("rel-1", "rg-1")],
716            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
717        }
718    )
719    ctrl = _mb_evidence_ctrl(musicbrainz)
720    base = _library_album(barcodes=(BASE_BARCODE, THIRD_BARCODE))
721    compare = _album("s1", "spotify_1", barcodes=(OTHER_BARCODE,))
722
723    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.INSUFFICIENT
724
725
726# ---------------------------------------------------------------------------
727# library insert path
728# ---------------------------------------------------------------------------
729
730
731@dataclass
732class _InsertHarness:
733    """A controller under test on the library insert path, with its IO boundaries mocked."""
734
735    ctrl: AlbumsController
736    get_provider: Mock
737    get_provider_album_tracks: AsyncMock
738
739    def album_track_calls(self) -> list[str]:
740        """Return the album item ids passed to each provider album-track lookup."""
741        return [call.args[0] for call in self.get_provider_album_tracks.await_args_list]
742
743
744def _library_candidate(
745    *, version: str = "", year: int | None = 2022, barcodes: Sequence[str] = ()
746) -> Album:
747    """Build the existing library album a provider album is compared against."""
748    return _album(
749        "1", "library", version=version, year=year, barcodes=barcodes, mappings=[BASE_MAPPING]
750    )
751
752
753@contextmanager
754def _insert_harness(
755    *,
756    candidates: Sequence[Album],
757    provider_album_tracks: dict[str, list[Track] | Exception] | None = None,
758    musicbrainz: Mock | None = None,
759    incoming_instance: str = "spotify_1",
760) -> Iterator[_InsertHarness]:
761    """
762    Yield an AlbumsController whose library lookups return the given name candidates.
763
764    :param candidates: Library albums returned by the normalized-name lookup.
765    :param provider_album_tracks: Provider album tracks (or an error to raise) keyed by
766        album item id, for both the candidate and the base album's mappings.
767    :param musicbrainz: Optional mock MusicBrainz provider.
768    :param incoming_instance: Instance id the incoming album's provider reports as its own;
769        a value other than "spotify_1" stands in for a same-domain fallback.
770    """
771    album_tracks = provider_album_tracks or {}
772
773    async def _album_tracks(item_id: str, *_rest: object) -> list[Track]:
774        result = album_tracks.get(item_id, [])
775        if isinstance(result, Exception):
776            raise result
777        return list(result)
778
779    # a single recorder backs both the base mapping lookup (_get_provider_album_tracks)
780    # and the candidate lookup (provider.get_album_tracks), so their calls stay ordered
781    get_provider_album_tracks = AsyncMock(side_effect=_album_tracks)
782    incoming_provider = _loaded_provider(incoming_instance)
783    incoming_provider.get_album_tracks = get_provider_album_tracks
784    registry = {"tidal_1": _loaded_provider("tidal_1"), "spotify_1": incoming_provider}
785
786    def _get_provider(instance: str, **_kwargs: object) -> object:
787        if instance == "musicbrainz":
788            return musicbrainz
789        return registry.get(instance)
790
791    mass = Mock()
792    mass.get_provider = Mock(side_effect=_get_provider)
793    ctrl = AlbumsController.__new__(AlbumsController)
794    ctrl.logger = logging.getLogger("test.albums.insert")
795    ctrl.mass = mass
796    ctrl.db_table = "albums"
797    with patch.multiple(
798        ctrl,
799        get_library_item_by_prov_id=AsyncMock(return_value=None),
800        get_library_item_by_prov_mappings=AsyncMock(return_value=None),
801        get_library_items_by_external_id=AsyncMock(return_value=[]),
802        get_library_items_by_query=AsyncMock(return_value=list(candidates)),
803        _get_provider_album_tracks=get_provider_album_tracks,
804        # the insert path resolves an album from the library, it never searches a provider
805        search=AsyncMock(side_effect=AssertionError("search during insert")),
806        get_provider_item=AsyncMock(side_effect=AssertionError("provider fetch during insert")),
807    ):
808        yield _InsertHarness(ctrl, mass.get_provider, get_provider_album_tracks)
809
810
811async def test_insert_match_is_io_free_without_candidates() -> None:
812    """An album with no same-name library candidate never reaches a provider."""
813    with _insert_harness(candidates=[]) as harness:
814        item = _album("a1", "spotify_1", barcodes=[BASE_BARCODE])
815
816        assert await harness.ctrl._get_library_item_by_match(item) is None
817
818    harness.get_provider.assert_not_called()
819    assert harness.album_track_calls() == []
820
821
822async def test_insert_match_does_not_escalate_an_unambiguous_candidate() -> None:
823    """A candidate the albums' own metadata already decides is never escalated."""
824    # a recording-changing edition conflict is decisive on metadata alone
825    with _insert_harness(candidates=[_library_candidate(version="Live")]) as harness:
826        item = _album("a1", "spotify_1", version="Deluxe")
827
828        assert await harness.ctrl._get_library_item_by_match(item) is None
829
830    harness.get_provider.assert_not_called()
831    assert harness.album_track_calls() == []
832
833
834async def test_insert_match_escalates_ambiguous_candidate_to_existing_album() -> None:
835    """An ambiguous edition with a matching tracklist is linked, not inserted again."""
836    with _insert_harness(
837        candidates=[_library_candidate()],
838        # year drift alone is ambiguous; identical tracklists resolve it
839        provider_album_tracks={"base-prov": _tracklist(10), "a1": _tracklist(10)},
840    ) as harness:
841        item = _album("a1", "spotify_1", year=2023)
842
843        assert await harness.ctrl._get_library_item_by_match(item) == 1
844
845    assert harness.album_track_calls() == ["base-prov", "a1"]
846
847
848async def test_insert_match_keeps_ambiguous_candidate_with_conflicting_tracklist() -> None:
849    """An ambiguous edition whose tracklist conflicts stays a separate library album."""
850    with _insert_harness(
851        candidates=[_library_candidate()],
852        provider_album_tracks={"base-prov": _tracklist(10), "a1": _tracklist(14)},
853    ) as harness:
854        item = _album("a1", "spotify_1", year=2023)
855
856        assert await harness.ctrl._get_library_item_by_match(item) is None
857
858    # the tracklists must have been the thing that rejected it, not a skipped escalation
859    assert harness.album_track_calls() == ["base-prov", "a1"]
860
861
862async def test_insert_match_escalates_each_candidate_against_its_own_tracklist() -> None:
863    """Every ambiguous candidate is fingerprinted against its own base tracklist."""
864    first = _library_candidate()
865    first.item_id = "7"
866    first.provider_mappings = {
867        ProviderMapping(item_id="other-prov", provider_domain="tidal", provider_instance="tidal_1")
868    }
869    with _insert_harness(
870        candidates=[first, _library_candidate()],
871        # only the second candidate shares the incoming tracklist
872        provider_album_tracks={
873            "other-prov": _tracklist(14),
874            "base-prov": _tracklist(10),
875            "a1": _tracklist(10),
876        },
877    ) as harness:
878        item = _album("a1", "spotify_1", year=2023)
879
880        assert await harness.ctrl._get_library_item_by_match(item) == 1
881
882    assert harness.album_track_calls() == ["other-prov", "a1", "base-prov", "a1"]
883
884
885async def test_insert_match_falls_through_to_musicbrainz_on_transient_error() -> None:
886    """A tracklist that is temporarily unavailable leaves the decision to MusicBrainz."""
887    with _insert_harness(
888        candidates=[_library_candidate(barcodes=(BASE_BARCODE,))],
889        provider_album_tracks={"base-prov": RetriesExhausted("provider unreachable")},
890        musicbrainz=_mb(
891            **{
892                BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
893                OTHER_BARCODE: [_mb_release("rel-1", "rg-1")],
894            }
895        ),
896    ) as harness:
897        item = _album("a1", "spotify_1", year=2023, barcodes=(OTHER_BARCODE,))
898
899        assert await harness.ctrl._get_library_item_by_match(item) == 1
900
901
902async def test_insert_match_abstains_when_musicbrainz_is_not_loaded() -> None:
903    """Without MusicBrainz an unresolved album is added rather than guessed."""
904    with _insert_harness(candidates=[_library_candidate(barcodes=(BASE_BARCODE,))]) as harness:
905        item = _album("a1", "spotify_1", year=2023, barcodes=(OTHER_BARCODE,))
906
907        assert await harness.ctrl._get_library_item_by_match(item) is None
908
909
910async def test_insert_match_escalates_to_musicbrainz_when_tracklists_are_absent() -> None:
911    """With no usable tracklist, a shared MusicBrainz release still links the album."""
912    with _insert_harness(
913        # no tracklist is available on either side, so only MusicBrainz can decide
914        candidates=[_library_candidate(barcodes=(BASE_BARCODE,))],
915        musicbrainz=_mb(
916            **{
917                BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
918                OTHER_BARCODE: [_mb_release("rel-1", "rg-1")],
919            }
920        ),
921    ) as harness:
922        item = _album("a1", "spotify_1", year=2023, barcodes=(OTHER_BARCODE,))
923
924        assert await harness.ctrl._get_library_item_by_match(item) == 1
925
926
927async def test_insert_match_skips_escalation_when_exact_instance_not_loaded() -> None:
928    """A same-domain fallback is never fingerprinted as the incoming album's provider."""
929    with _insert_harness(
930        candidates=[_library_candidate()],
931        provider_album_tracks={"base-prov": _tracklist(10), "a1": _tracklist(10)},
932        # the loaded instance reports a different id: a second account of the same domain
933        incoming_instance="spotify_2",
934    ) as harness:
935        item = _album("a1", "spotify_1", year=2023)
936
937        assert await harness.ctrl._get_library_item_by_match(item) is None
938
939    assert harness.album_track_calls() == []
940