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