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