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_search_string_combines_artist_and_album_name() -> None:
277    """The provider search is made on the album artist and the album name."""
278    with _harness(search_results=[], provider_items={}) as harness:
279        await harness.match(_library_album())
280
281    harness.search.assert_awaited_once_with("Sigur Rós - ( )", "spotify_1")
282
283
284async def test_album_without_artists_searches_on_name_only() -> None:
285    """An album that has no artists is still searchable instead of raising."""
286    base = _library_album()
287    base.artists = UniqueList()
288    with _harness(search_results=[], provider_items={}) as harness:
289        matches = await harness.match(base)
290
291    assert matches == []
292    harness.search.assert_awaited_once_with("( )", "spotify_1")
293
294
295async def test_insufficient_search_result_proceeds_to_one_full_fetch() -> None:
296    """An ambiguous search result is confirmed against exactly one full provider album."""
297    base = _library_album(external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
298    sparse = _album("s1", "spotify_1", version="Remaster")
299    full = _album("s1", "spotify_1", external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
300    with _harness(search_results=[sparse], provider_items={"s1": full}) as harness:
301        matches = await harness.match(base)
302
303    assert [mapping.item_id for mapping in matches] == ["s1"]
304    harness.get_provider_item.assert_awaited_once()
305
306
307async def test_full_item_match_uses_no_track_or_musicbrainz_calls() -> None:
308    """A full album that matches on metadata never fetches tracks or hits MusicBrainz."""
309    musicbrainz = _mb()
310    base = _library_album(external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
311    sparse = _album("s1", "spotify_1", version="Remaster")
312    full = _album("s1", "spotify_1", external_ids={(ExternalID.MB_ALBUM, MB_ALBUM_ID)})
313    with _harness(
314        search_results=[sparse], provider_items={"s1": full}, musicbrainz=musicbrainz
315    ) as harness:
316        matches = await harness.match(base)
317
318    assert [mapping.item_id for mapping in matches] == ["s1"]
319    harness.get_provider_album_tracks.assert_not_awaited()
320    musicbrainz.get_releases_by_barcode.assert_not_awaited()
321
322
323# ---------------------------------------------------------------------------
324# track-fingerprint resolution
325# ---------------------------------------------------------------------------
326
327
328async def test_ambiguous_albums_resolve_via_track_fingerprints() -> None:
329    """The 14-track remaster vs deluxe-remaster case matches once tracklists agree."""
330    musicbrainz = _mb()
331    base = _library_album(version="2022 Remaster")
332    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
333    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
334    with _harness(
335        search_results=[sparse],
336        provider_items={"s1": full},
337        provider_album_tracks={"base-prov": _tracklist(14), "s1": _tracklist(14)},
338        musicbrainz=musicbrainz,
339    ) as harness:
340        matches = await harness.match(base)
341
342    assert [mapping.item_id for mapping in matches] == ["s1"]
343    # base tracklist comes from the existing provider mapping, candidate from the match target
344    assert harness.album_track_calls() == ["base-prov", "s1"]
345    # a decisive fingerprint match must not fall through to MusicBrainz
346    musicbrainz.get_releases_by_barcode.assert_not_awaited()
347
348
349async def test_different_track_counts_do_not_map() -> None:
350    """A complete 8-track album and a 14-track edition are never merged by fingerprints."""
351    base = _library_album(version="2022 Remaster")
352    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
353    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
354    with _harness(
355        search_results=[sparse],
356        provider_items={"s1": full},
357        provider_album_tracks={"base-prov": _tracklist(8), "s1": _tracklist(14)},
358    ) as harness:
359        assert await harness.match(base) == []
360
361
362async def test_conflicting_isrc_fingerprints_do_not_map() -> None:
363    """Same-length tracklists with conflicting ISRCs are a different recording."""
364    base = _library_album(version="2022 Remaster")
365    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
366    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
367    with _harness(
368        search_results=[sparse],
369        provider_items={"s1": full},
370        provider_album_tracks={
371            "base-prov": _tracklist(14, isrc_prefix="USRC17607"),
372            "s1": _tracklist(14, isrc_prefix="USRC28718"),
373        },
374    ) as harness:
375        assert await harness.match(base) == []
376
377
378async def test_base_tracks_fetched_once_across_candidates() -> None:
379    """The base tracklist is fetched once even when several candidates need it."""
380    base = _library_album(version="2022 Remaster")
381    search_results = [
382        _album(f"s{index}", "spotify_1", version="Deluxe 2022 Remaster") for index in range(1, 4)
383    ]
384    provider_items = {album.item_id: album for album in search_results}
385    with _harness(
386        search_results=search_results,
387        provider_items=provider_items,
388        # every candidate stays ambiguous (no tracks to compare against)
389        provider_album_tracks={"base-prov": _tracklist(14)},
390    ) as harness:
391        await harness.match(base)
392
393    # the base mapping is only fetched once, regardless of the number of candidates
394    assert harness.album_track_calls().count("base-prov") == 1
395
396
397async def test_base_tracks_use_first_trustworthy_loaded_mapping() -> None:
398    """Base resolution deterministically skips unloaded/untrustworthy mappings."""
399    base = _library_album(
400        version="2022 Remaster",
401        mappings=[
402            ProviderMapping(item_id="a", provider_domain="aaa", provider_instance="aaa_1"),
403            ProviderMapping(item_id="b", provider_domain="bbb", provider_instance="bbb_1"),
404            ProviderMapping(item_id="c", provider_domain="ccc", provider_instance="ccc_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            # "a" is on an unloaded instance and must never be fetched
414            "a": _tracklist(14),
415            # "b" is loaded but its positions are untrustworthy (no disc numbers)
416            "b": _tracklist(14, disc_number=0),
417            "c": _tracklist(14),
418            "s1": _tracklist(14),
419        },
420        loaded_instances=("bbb_1", "ccc_1", "tidal_1"),
421    ) as harness:
422        matches = await harness.match(base)
423
424    assert [mapping.item_id for mapping in matches] == ["s1"]
425    # unloaded "a" skipped before any fetch; "b" fetched but rejected; "c" used
426    assert harness.album_track_calls() == ["b", "c", "s1"]
427
428
429async def test_base_mapping_not_found_is_skipped() -> None:
430    """A base mapping whose tracklist lookup 404s is skipped for the next mapping."""
431    base = _library_album(
432        version="2022 Remaster",
433        mappings=[
434            ProviderMapping(item_id="gone", provider_domain="aaa", provider_instance="aaa_1"),
435            ProviderMapping(item_id="ok", provider_domain="bbb", provider_instance="bbb_1"),
436        ],
437    )
438    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
439    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
440    with _harness(
441        search_results=[sparse],
442        provider_items={"s1": full},
443        provider_album_tracks={
444            "gone": MediaNotFoundError("gone"),
445            "ok": _tracklist(14),
446            "s1": _tracklist(14),
447        },
448        loaded_instances=("aaa_1", "bbb_1"),
449    ) as harness:
450        matches = await harness.match(base)
451
452    assert [mapping.item_id for mapping in matches] == ["s1"]
453    assert harness.album_track_calls() == ["gone", "ok", "s1"]
454
455
456async def test_base_mapping_transient_error_is_skipped() -> None:
457    """A transient base tracklist failure skips to the next mapping."""
458    base = _library_album(
459        version="2022 Remaster",
460        mappings=[
461            ProviderMapping(item_id="flaky", provider_domain="aaa", provider_instance="aaa_1"),
462            ProviderMapping(item_id="ok", provider_domain="bbb", provider_instance="bbb_1"),
463        ],
464    )
465    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
466    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
467    with _harness(
468        search_results=[sparse],
469        provider_items={"s1": full},
470        provider_album_tracks={
471            "flaky": RetriesExhausted("flaky"),
472            "ok": _tracklist(14),
473            "s1": _tracklist(14),
474        },
475        loaded_instances=("aaa_1", "bbb_1"),
476    ) as harness:
477        matches = await harness.match(base)
478
479    assert [mapping.item_id for mapping in matches] == ["s1"]
480    assert harness.album_track_calls() == ["flaky", "ok", "s1"]
481
482
483async def test_base_mapping_skipped_when_exact_instance_not_loaded() -> None:
484    """A mapping whose exact instance is not loaded is skipped, never a same-domain fallback."""
485    base = _library_album(
486        version="2022 Remaster",
487        mappings=[
488            ProviderMapping(item_id="wrong", provider_domain="aaa", provider_instance="aaa_1"),
489            ProviderMapping(item_id="ok", provider_domain="bbb", provider_instance="bbb_1"),
490        ],
491    )
492    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
493    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
494    with _harness(
495        search_results=[sparse],
496        provider_items={"s1": full},
497        provider_album_tracks={"wrong": _tracklist(14), "ok": _tracklist(14), "s1": _tracklist(14)},
498        loaded_instances=("bbb_1",),
499        # aaa_1 resolves (same-domain fallback) to a different instance, so it must be skipped
500        provider_registry={"aaa_1": _loaded_provider("other_1")},
501    ) as harness:
502        matches = await harness.match(base)
503
504    assert [mapping.item_id for mapping in matches] == ["s1"]
505    # "wrong" is never fetched because its exact instance isn't the one that resolved
506    assert harness.album_track_calls() == ["ok", "s1"]
507
508
509async def test_candidate_tracklist_not_found_falls_through_to_musicbrainz() -> None:
510    """A candidate tracklist that 404s is treated as absent so MusicBrainz can decide."""
511    musicbrainz = _mb(
512        **{
513            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
514            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
515        }
516    )
517    base = _library_album(version="2022 Remaster", barcodes=[BASE_BARCODE])
518    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
519    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
520    with _harness(
521        search_results=[sparse],
522        provider_items={"s1": full},
523        provider_album_tracks={"base-prov": _tracklist(14), "s1": MediaNotFoundError("s1")},
524        musicbrainz=musicbrainz,
525    ) as harness:
526        matches = await harness.match(base)
527
528    # MusicBrainz was consulted despite the tracklist failure and its verdict
529    # (disjoint release groups) was applied
530    assert matches == []
531    musicbrainz.get_releases_by_barcode.assert_awaited()
532
533
534async def test_candidate_tracklist_transient_error_falls_through_to_musicbrainz() -> None:
535    """A transient candidate tracklist failure is treated as absent so MusicBrainz can decide."""
536    musicbrainz = _mb(
537        **{
538            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
539            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
540        }
541    )
542    base = _library_album(version="2022 Remaster", barcodes=[BASE_BARCODE])
543    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
544    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=[OTHER_BARCODE])
545    with _harness(
546        search_results=[sparse],
547        provider_items={"s1": full},
548        provider_album_tracks={"base-prov": _tracklist(14), "s1": TimeoutError()},
549        musicbrainz=musicbrainz,
550    ) as harness:
551        matches = await harness.match(base)
552
553    # MusicBrainz was consulted despite the tracklist failure and its verdict
554    # (disjoint release groups) was applied
555    assert matches == []
556    musicbrainz.get_releases_by_barcode.assert_awaited()
557
558
559async def test_candidate_tracklist_uses_matched_provider_not_same_domain_fallback() -> None:
560    """The candidate tracklist is fetched from the matched provider, never a same-domain fallback."""
561    # a second instance of the candidate's domain is registered in the loaded-provider
562    # registry; re-resolving the candidate through it would fingerprint against the wrong
563    # account and reject the correct match
564    fallback = _loaded_provider("spotify_other")
565    fallback.get_album_tracks = AsyncMock(return_value=_tracklist(14, isrc_prefix="USRC28718"))
566    base = _library_album(version="2022 Remaster")
567    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
568    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster")
569    with _harness(
570        search_results=[sparse],
571        provider_items={"s1": full},
572        provider_album_tracks={"base-prov": _tracklist(14)},
573        provider_registry={"spotify_1": fallback},
574    ) as harness:
575        # the matched provider itself returns the correct, agreeing tracklist
576        harness.provider.get_album_tracks = AsyncMock(return_value=_tracklist(14))
577        matches = await harness.match(base)
578
579    assert [mapping.item_id for mapping in matches] == ["s1"]
580    harness.provider.get_album_tracks.assert_awaited_once_with("s1")
581    # the same-domain fallback is never consulted and the candidate domain is never re-resolved
582    fallback.get_album_tracks.assert_not_awaited()
583    assert all(call.args[0] != "spotify_1" for call in harness.get_provider.call_args_list)
584
585
586# ---------------------------------------------------------------------------
587# MusicBrainz last-resort evidence
588# ---------------------------------------------------------------------------
589
590
591@contextmanager
592def _mb_harness(
593    musicbrainz: Mock | None,
594    *,
595    base_barcodes: Sequence[str] = (BASE_BARCODE,),
596    compare_barcodes: Sequence[str] = (OTHER_BARCODE,),
597) -> Iterator[tuple[_Harness, Album]]:
598    """Yield a harness plus base album whose only candidate is ambiguous past fingerprints."""
599    base = _library_album(version="2022 Remaster", barcodes=base_barcodes)
600    sparse = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=compare_barcodes)
601    full = _album("s1", "spotify_1", version="Deluxe 2022 Remaster", barcodes=compare_barcodes)
602    with _harness(
603        search_results=[sparse],
604        provider_items={"s1": full},
605        # base tracklist present but candidate has none: fingerprint stays inconclusive
606        provider_album_tracks={"base-prov": _tracklist(14), "s1": []},
607        musicbrainz=musicbrainz,
608    ) as harness:
609        yield harness, base
610
611
612async def test_shared_barcode_resolves_ambiguity_without_musicbrainz() -> None:
613    """A shared barcode resolves an ambiguous edition wording without any lookups."""
614    musicbrainz = _mb()
615    with _mb_harness(musicbrainz, compare_barcodes=(BASE_BARCODE,)) as (harness, base):
616        matches = await harness.match(base)
617
618    assert [mapping.item_id for mapping in matches] == ["s1"]
619    musicbrainz.get_releases_by_barcode.assert_not_awaited()
620
621
622async def test_musicbrainz_multiple_releases_per_barcode_abstains() -> None:
623    """A barcode resolving to several releases is ambiguous and must not match."""
624    musicbrainz = _mb(
625        **{
626            BASE_BARCODE: [_mb_release("rel-1", "rg-1"), _mb_release("rel-2", "rg-1")],
627            OTHER_BARCODE: [_mb_release("rel-3", "rg-1")],
628        }
629    )
630    with _mb_harness(musicbrainz) as (harness, base):
631        assert await harness.match(base) == []
632
633
634async def test_musicbrainz_multiple_barcodes_are_all_considered() -> None:
635    """Every canonical barcode on each album is used, not just the first."""
636    musicbrainz = _mb(
637        **{
638            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
639            THIRD_BARCODE: [_mb_release("rel-9", "rg-9")],
640            OTHER_BARCODE: [_mb_release("rel-2", "rg-9")],
641        }
642    )
643    with _mb_harness(
644        musicbrainz, base_barcodes=(BASE_BARCODE, THIRD_BARCODE), compare_barcodes=(OTHER_BARCODE,)
645    ) as (harness, base):
646        assert await harness.match(base) == []
647    queried = {call.args[0] for call in musicbrainz.get_releases_by_barcode.await_args_list}
648    assert queried == {BASE_BARCODE, OTHER_BARCODE, THIRD_BARCODE}
649
650
651async def test_musicbrainz_shared_release_group_alone_does_not_match() -> None:
652    """Different specific releases in one release group must not identify an edition."""
653    musicbrainz = _mb(
654        **{
655            BASE_BARCODE: [_mb_release("rel-1", "rg-9")],
656            OTHER_BARCODE: [_mb_release("rel-2", "rg-9")],
657        }
658    )
659    with _mb_harness(musicbrainz, compare_barcodes=(OTHER_BARCODE,)) as (harness, base):
660        assert await harness.match(base) == []
661
662
663async def test_musicbrainz_disjoint_release_groups_do_not_match() -> None:
664    """Barcodes in entirely different release groups are negative evidence."""
665    musicbrainz = _mb(
666        **{
667            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
668            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
669        }
670    )
671    with _mb_harness(musicbrainz, compare_barcodes=(OTHER_BARCODE,)) as (harness, base):
672        assert await harness.match(base) == []
673
674
675async def test_musicbrainz_unresolved_barcode_abstains() -> None:
676    """A barcode MusicBrainz cannot resolve abstains instead of guessing."""
677    with _mb_harness(_mb()) as (harness, base):
678        assert await harness.match(base) == []
679
680
681async def test_musicbrainz_transport_error_abstains() -> None:
682    """A MusicBrainz outage abstains rather than aborting the whole match."""
683    musicbrainz = Mock()
684    musicbrainz.get_releases_by_barcode = AsyncMock(side_effect=TimeoutError())
685    with _mb_harness(musicbrainz) as (harness, base):
686        assert await harness.match(base) == []
687
688
689async def test_musicbrainz_skipped_without_a_configured_provider() -> None:
690    """With no MusicBrainz provider configured the match simply abstains."""
691    with _mb_harness(None) as (harness, base):
692        assert await harness.match(base) == []
693
694
695def _mb_evidence_ctrl(musicbrainz: Mock) -> AlbumsController:
696    """Return a bare controller wired to a mock MusicBrainz provider."""
697    mass = Mock()
698    mass.get_provider = Mock(
699        side_effect=lambda domain: musicbrainz if domain == "musicbrainz" else None
700    )
701    ctrl = AlbumsController.__new__(AlbumsController)
702    ctrl.logger = logging.getLogger("test.albums.mb")
703    ctrl.mass = mass
704    return ctrl
705
706
707async def test_musicbrainz_evidence_shared_single_release_matches() -> None:
708    """A shared barcode resolving to exactly one specific release is positive evidence."""
709    musicbrainz = _mb(**{BASE_BARCODE: [_mb_release("rel-1", "rg-1")]})
710    ctrl = _mb_evidence_ctrl(musicbrainz)
711    base = _library_album(barcodes=(BASE_BARCODE,))
712    compare = _album("s1", "spotify_1", barcodes=(BASE_BARCODE,))
713
714    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.MATCH
715
716
717async def test_musicbrainz_evidence_ambiguous_barcode_is_not_release_identity() -> None:
718    """A shared barcode resolving to several releases never identifies a specific release."""
719    musicbrainz = _mb(
720        **{BASE_BARCODE: [_mb_release("rel-1", "rg-1"), _mb_release("rel-2", "rg-1")]}
721    )
722    ctrl = _mb_evidence_ctrl(musicbrainz)
723    base = _library_album(barcodes=(BASE_BARCODE,))
724    compare = _album("s1", "spotify_1", barcodes=(BASE_BARCODE,))
725
726    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.INSUFFICIENT
727
728
729async def test_musicbrainz_evidence_all_resolved_disjoint_groups_reject() -> None:
730    """Disjoint release groups are negative evidence when every barcode resolved."""
731    musicbrainz = _mb(
732        **{
733            BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
734            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
735        }
736    )
737    ctrl = _mb_evidence_ctrl(musicbrainz)
738    base = _library_album(barcodes=(BASE_BARCODE,))
739    compare = _album("s1", "spotify_1", barcodes=(OTHER_BARCODE,))
740
741    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.NO_MATCH
742
743
744async def test_musicbrainz_evidence_unresolved_barcode_blocks_disjoint_rejection() -> None:
745    """An unresolved barcode leaves the group sets incomplete, so it cannot reject."""
746    musicbrainz = _mb(
747        **{
748            # BASE_BARCODE stays unresolved ([]); THIRD_BARCODE resolves to a distinct group
749            THIRD_BARCODE: [_mb_release("rel-1", "rg-1")],
750            OTHER_BARCODE: [_mb_release("rel-2", "rg-2")],
751        }
752    )
753    ctrl = _mb_evidence_ctrl(musicbrainz)
754    base = _library_album(barcodes=(BASE_BARCODE, THIRD_BARCODE))
755    compare = _album("s1", "spotify_1", barcodes=(OTHER_BARCODE,))
756
757    assert await ctrl._musicbrainz_album_evidence(base, compare) == AlbumMatchEvidence.INSUFFICIENT
758
759
760# ---------------------------------------------------------------------------
761# library insert path
762# ---------------------------------------------------------------------------
763
764
765@dataclass
766class _InsertHarness:
767    """A controller under test on the library insert path, with its IO boundaries mocked."""
768
769    ctrl: AlbumsController
770    get_provider: Mock
771    get_provider_album_tracks: AsyncMock
772    get_library_items_by_query: AsyncMock
773
774    def album_track_calls(self) -> list[str]:
775        """Return the album item ids passed to each provider album-track lookup."""
776        return [call.args[0] for call in self.get_provider_album_tracks.await_args_list]
777
778
779def _library_candidate(
780    *, version: str = "", year: int | None = 2022, barcodes: Sequence[str] = ()
781) -> Album:
782    """Build the existing library album a provider album is compared against."""
783    return _album(
784        "1", "library", version=version, year=year, barcodes=barcodes, mappings=[BASE_MAPPING]
785    )
786
787
788@contextmanager
789def _insert_harness(
790    *,
791    candidates: Sequence[Album],
792    provider_album_tracks: dict[str, list[Track] | Exception] | None = None,
793    musicbrainz: Mock | None = None,
794    incoming_instance: str = "spotify_1",
795) -> Iterator[_InsertHarness]:
796    """
797    Yield an AlbumsController whose library lookups return the given name candidates.
798
799    :param candidates: Library albums returned by the normalized-name lookup.
800    :param provider_album_tracks: Provider album tracks (or an error to raise) keyed by
801        album item id, for both the candidate and the base album's mappings.
802    :param musicbrainz: Optional mock MusicBrainz provider.
803    :param incoming_instance: Instance id the incoming album's provider reports as its own;
804        a value other than "spotify_1" stands in for a same-domain fallback.
805    """
806    album_tracks = provider_album_tracks or {}
807
808    async def _album_tracks(item_id: str, *_rest: object) -> list[Track]:
809        result = album_tracks.get(item_id, [])
810        if isinstance(result, Exception):
811            raise result
812        return list(result)
813
814    # a single recorder backs both the base mapping lookup (_get_provider_album_tracks)
815    # and the candidate lookup (provider.get_album_tracks), so their calls stay ordered
816    get_provider_album_tracks = AsyncMock(side_effect=_album_tracks)
817    get_library_items_by_query = AsyncMock(return_value=list(candidates))
818    incoming_provider = _loaded_provider(incoming_instance)
819    incoming_provider.get_album_tracks = get_provider_album_tracks
820    registry = {"tidal_1": _loaded_provider("tidal_1"), "spotify_1": incoming_provider}
821
822    def _get_provider(instance: str, **_kwargs: object) -> object:
823        if instance == "musicbrainz":
824            return musicbrainz
825        return registry.get(instance)
826
827    mass = Mock()
828    mass.get_provider = Mock(side_effect=_get_provider)
829    ctrl = AlbumsController.__new__(AlbumsController)
830    ctrl.logger = logging.getLogger("test.albums.insert")
831    ctrl.mass = mass
832    ctrl.db_table = "albums"
833    with patch.multiple(
834        ctrl,
835        get_library_item_by_prov_id=AsyncMock(return_value=None),
836        get_library_item_by_prov_mappings=AsyncMock(return_value=None),
837        get_library_items_by_external_id=AsyncMock(return_value=[]),
838        get_library_items_by_query=get_library_items_by_query,
839        _get_provider_album_tracks=get_provider_album_tracks,
840        # the insert path resolves an album from the library, it never searches a provider
841        search=AsyncMock(side_effect=AssertionError("search during insert")),
842        get_provider_item=AsyncMock(side_effect=AssertionError("provider fetch during insert")),
843    ):
844        yield _InsertHarness(
845            ctrl, mass.get_provider, get_provider_album_tracks, get_library_items_by_query
846        )
847
848
849async def test_insert_match_is_io_free_without_candidates() -> None:
850    """An album with no same-name library candidate never reaches a provider."""
851    with _insert_harness(candidates=[]) as harness:
852        item = _album("a1", "spotify_1", barcodes=[BASE_BARCODE])
853
854        assert await harness.ctrl._get_library_item_by_match(item) is None
855
856    harness.get_provider.assert_not_called()
857    assert harness.album_track_calls() == []
858
859
860async def test_insert_match_looks_up_the_retail_suffix_spellings() -> None:
861    """An album is sought under its plain title and every spelled-out retail suffix."""
862
863    async def searched_names(harness: _InsertHarness, name: str) -> list[str]:
864        assert (
865            await harness.ctrl._get_library_item_by_match(_album("a1", "spotify_1", name=name))
866            is None
867        )
868        params = harness.get_library_items_by_query.await_args_list[-1].kwargs["extra_query_params"]
869        return list(params["search_names"])
870
871    expected = ["stargazing", "stargazingep", "stargazingsingle"]
872    with _insert_harness(candidates=[]) as harness:
873        assert await searched_names(harness, "Stargazing") == expected
874        for spelling in RETAIL_SUFFIX_NAMES:
875            assert await searched_names(harness, spelling) == expected, spelling
876
877
878@pytest.mark.parametrize("suffixed_name", RETAIL_SUFFIX_NAMES)
879async def test_insert_match_links_a_spelled_out_retail_suffix_to_the_plain_title(
880    mass: MusicAssistant, suffixed_name: str
881) -> None:
882    """A title naming the format joins the existing plain-title row instead of duplicating it."""
883    artist = await mass.music.artists.add_item_to_library(
884        Artist(
885            item_id="0",
886            provider="library",
887            name="Kygo",
888            provider_mappings={
889                ProviderMapping(
890                    item_id="artist", provider_domain="spotify", provider_instance="spotify_1"
891                )
892            },
893        )
894    )
895    plain = await mass.music.albums.add_item_to_library(
896        Album(
897            item_id="0",
898            provider="library",
899            name="Stargazing",
900            artists=UniqueList([artist]),
901            provider_mappings={
902                ProviderMapping(
903                    item_id="plain", provider_domain="spotify", provider_instance="spotify_1"
904                )
905            },
906        )
907    )
908    suffixed = await mass.music.albums.add_item_to_library(
909        Album(
910            item_id="suffixed",
911            provider="apple_music_1",
912            name=suffixed_name,
913            artists=UniqueList([artist]),
914            provider_mappings={
915                ProviderMapping(
916                    item_id="suffixed",
917                    provider_domain="apple_music",
918                    provider_instance="apple_music_1",
919                )
920            },
921        )
922    )
923
924    assert suffixed.item_id == plain.item_id
925
926
927async def test_insert_match_keeps_an_ep_and_a_single_of_the_same_name_apart(
928    mass: MusicAssistant,
929) -> None:
930    """An EP and a single sharing a base title are separate releases, not one library row."""
931    artist = await mass.music.artists.add_item_to_library(
932        Artist(
933            item_id="0",
934            provider="library",
935            name="Kygo",
936            provider_mappings={
937                ProviderMapping(
938                    item_id="artist", provider_domain="spotify", provider_instance="spotify_1"
939                )
940            },
941        )
942    )
943
944    async def add(name: str, instance: str) -> Album:
945        return await mass.music.albums.add_item_to_library(
946            Album(
947                item_id=f"{instance}-item",
948                provider=instance,
949                name=name,
950                artists=UniqueList([artist]),
951                provider_mappings={
952                    ProviderMapping(
953                        item_id=f"{instance}-item",
954                        provider_domain=instance.rsplit("_", 1)[0],
955                        provider_instance=instance,
956                    )
957                },
958            )
959        )
960
961    ep = await add("Stargazing - EP", "apple_music_1")
962    single = await add("Stargazing - Single", "tidal_1")
963    assert ep.item_id != single.item_id
964
965    # the plain spelling still joins one of them rather than becoming a third row
966    plain = await add("Stargazing", "spotify_1")
967    assert plain.item_id in {ep.item_id, single.item_id}
968
969
970async def test_insert_match_does_not_escalate_an_unambiguous_candidate() -> None:
971    """A candidate the albums' own metadata already decides is never escalated."""
972    # a recording-changing edition conflict is decisive on metadata alone
973    with _insert_harness(candidates=[_library_candidate(version="Live")]) as harness:
974        item = _album("a1", "spotify_1", version="Deluxe")
975
976        assert await harness.ctrl._get_library_item_by_match(item) is None
977
978    harness.get_provider.assert_not_called()
979    assert harness.album_track_calls() == []
980
981
982async def test_insert_match_escalates_ambiguous_candidate_to_existing_album() -> None:
983    """An ambiguous edition with a matching tracklist is linked, not inserted again."""
984    with _insert_harness(
985        candidates=[_library_candidate()],
986        # year drift alone is ambiguous; identical tracklists resolve it
987        provider_album_tracks={"base-prov": _tracklist(10), "a1": _tracklist(10)},
988    ) as harness:
989        item = _album("a1", "spotify_1", year=2023)
990
991        assert await harness.ctrl._get_library_item_by_match(item) == 1
992
993    assert harness.album_track_calls() == ["base-prov", "a1"]
994
995
996async def test_insert_match_keeps_ambiguous_candidate_with_conflicting_tracklist() -> None:
997    """An ambiguous edition whose tracklist conflicts stays a separate library album."""
998    with _insert_harness(
999        candidates=[_library_candidate()],
1000        provider_album_tracks={"base-prov": _tracklist(10), "a1": _tracklist(14)},
1001    ) as harness:
1002        item = _album("a1", "spotify_1", year=2023)
1003
1004        assert await harness.ctrl._get_library_item_by_match(item) is None
1005
1006    # the tracklists must have been the thing that rejected it, not a skipped escalation
1007    assert harness.album_track_calls() == ["base-prov", "a1"]
1008
1009
1010async def test_insert_match_escalates_each_candidate_against_its_own_tracklist() -> None:
1011    """Every ambiguous candidate is fingerprinted against its own base tracklist."""
1012    first = _library_candidate()
1013    first.item_id = "7"
1014    first.provider_mappings = {
1015        ProviderMapping(item_id="other-prov", provider_domain="tidal", provider_instance="tidal_1")
1016    }
1017    with _insert_harness(
1018        candidates=[first, _library_candidate()],
1019        # only the second candidate shares the incoming tracklist
1020        provider_album_tracks={
1021            "other-prov": _tracklist(14),
1022            "base-prov": _tracklist(10),
1023            "a1": _tracklist(10),
1024        },
1025    ) as harness:
1026        item = _album("a1", "spotify_1", year=2023)
1027
1028        assert await harness.ctrl._get_library_item_by_match(item) == 1
1029
1030    assert harness.album_track_calls() == ["other-prov", "a1", "base-prov", "a1"]
1031
1032
1033async def test_insert_match_falls_through_to_musicbrainz_on_transient_error() -> None:
1034    """A tracklist that is temporarily unavailable leaves the decision to MusicBrainz."""
1035    with _insert_harness(
1036        candidates=[_library_candidate(barcodes=(BASE_BARCODE,))],
1037        provider_album_tracks={"base-prov": RetriesExhausted("provider unreachable")},
1038        musicbrainz=_mb(
1039            **{
1040                BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
1041                OTHER_BARCODE: [_mb_release("rel-1", "rg-1")],
1042            }
1043        ),
1044    ) as harness:
1045        item = _album("a1", "spotify_1", year=2023, barcodes=(OTHER_BARCODE,))
1046
1047        assert await harness.ctrl._get_library_item_by_match(item) == 1
1048
1049
1050async def test_insert_match_abstains_when_musicbrainz_is_not_loaded() -> None:
1051    """Without MusicBrainz an unresolved album is added rather than guessed."""
1052    with _insert_harness(candidates=[_library_candidate(barcodes=(BASE_BARCODE,))]) as harness:
1053        item = _album("a1", "spotify_1", year=2023, barcodes=(OTHER_BARCODE,))
1054
1055        assert await harness.ctrl._get_library_item_by_match(item) is None
1056
1057
1058async def test_insert_match_escalates_to_musicbrainz_when_tracklists_are_absent() -> None:
1059    """With no usable tracklist, a shared MusicBrainz release still links the album."""
1060    with _insert_harness(
1061        # no tracklist is available on either side, so only MusicBrainz can decide
1062        candidates=[_library_candidate(barcodes=(BASE_BARCODE,))],
1063        musicbrainz=_mb(
1064            **{
1065                BASE_BARCODE: [_mb_release("rel-1", "rg-1")],
1066                OTHER_BARCODE: [_mb_release("rel-1", "rg-1")],
1067            }
1068        ),
1069    ) as harness:
1070        item = _album("a1", "spotify_1", year=2023, barcodes=(OTHER_BARCODE,))
1071
1072        assert await harness.ctrl._get_library_item_by_match(item) == 1
1073
1074
1075async def test_insert_match_skips_escalation_when_exact_instance_not_loaded() -> None:
1076    """A same-domain fallback is never fingerprinted as the incoming album's provider."""
1077    with _insert_harness(
1078        candidates=[_library_candidate()],
1079        provider_album_tracks={"base-prov": _tracklist(10), "a1": _tracklist(10)},
1080        # the loaded instance reports a different id: a second account of the same domain
1081        incoming_instance="spotify_2",
1082    ) as harness:
1083        item = _album("a1", "spotify_1", year=2023)
1084
1085        assert await harness.ctrl._get_library_item_by_match(item) is None
1086
1087    assert harness.album_track_calls() == []
1088