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