music-assistant-server

1"""Tests for ArtistsController provider matching (explicit, IO-capable enrichment)."""
2
3from __future__ import annotations
4
5import logging
6from contextlib import contextmanager
7from dataclasses import dataclass
8from typing import TYPE_CHECKING
9from unittest.mock import AsyncMock, Mock, patch
10
11from music_assistant_models.enums import ArtistType, ExternalID, MediaType
12from music_assistant_models.errors import MediaNotFoundError
13from music_assistant_models.media_items import (
14    Album,
15    Artist,
16    ItemMapping,
17    ProviderMapping,
18    Track,
19    UniqueList,
20)
21
22from music_assistant.controllers.music.media.artists import ArtistsController
23
24if TYPE_CHECKING:
25    from collections.abc import Iterator, Sequence
26
27MB_ARTIST_ID = "11111111-1111-1111-1111-111111111111"
28OTHER_MB_ARTIST_ID = "22222222-2222-2222-2222-222222222222"
29LIBRARY_ITEM_ID = "lib1"
30# the base library artist is linked to one existing (already-loaded) provider
31BASE_MAPPING = ProviderMapping(
32    item_id="base-prov", provider_domain="tidal", provider_instance="tidal_1"
33)
34CANDIDATE_MAPPING = ProviderMapping(
35    item_id="cand", provider_domain="spotify", provider_instance="spotify_1"
36)
37
38
39# ---------------------------------------------------------------------------
40# builders
41# ---------------------------------------------------------------------------
42
43
44def _artist_id(name: str) -> str:
45    """Return the provider item id used for an artist with the given name."""
46    return name.lower().replace(" ", "-")
47
48
49def _credit(name: str, provider: str) -> ItemMapping:
50    """Build the simplified artist credit as it appears on a media item."""
51    return ItemMapping(
52        media_type=MediaType.ARTIST,
53        item_id=_artist_id(name),
54        provider=provider,
55        name=name,
56    )
57
58
59def _library_artist(
60    *,
61    name: str = "Main Artist",
62    artist_type: ArtistType = ArtistType.SINGER,
63    external_ids: set[tuple[ExternalID, str]] | None = None,
64) -> Artist:
65    """Build the base (library) artist under match."""
66    return Artist(
67        item_id=LIBRARY_ITEM_ID,
68        provider="library",
69        name=name,
70        artist_type=artist_type,
71        external_ids=external_ids or set(),
72        provider_mappings={BASE_MAPPING},
73    )
74
75
76def _provider_artist(
77    *,
78    name: str = "Main Artist",
79    artist_type: ArtistType = ArtistType.SINGER,
80    external_ids: set[tuple[ExternalID, str]] | None = None,
81) -> Artist:
82    """Build the full provider artist returned for a credited candidate."""
83    return Artist(
84        item_id=_artist_id(name),
85        provider="spotify_1",
86        name=name,
87        artist_type=artist_type,
88        external_ids=external_ids or set(),
89        provider_mappings={CANDIDATE_MAPPING},
90    )
91
92
93def _track(
94    item_id: str,
95    provider: str,
96    *,
97    name: str = "Track One",
98    album_name: str = "Album X",
99    duration: int = 200,
100    artist_names: Sequence[str] = ("Main Artist",),
101    mappings: Sequence[ProviderMapping] | None = None,
102) -> Track:
103    """Build a track for the reference-track leg, with its artists credited by name."""
104    if mappings is None:
105        mappings = [
106            ProviderMapping(item_id=item_id, provider_domain=provider, provider_instance=provider)
107        ]
108    return Track(
109        item_id=item_id,
110        provider=provider,
111        name=name,
112        duration=duration,
113        disc_number=1,
114        track_number=1,
115        artists=UniqueList([_credit(artist_name, provider) for artist_name in artist_names]),
116        album=ItemMapping(
117            media_type=MediaType.ALBUM,
118            item_id=f"{item_id}-album",
119            provider=provider,
120            name=album_name,
121        ),
122        provider_mappings=set(mappings),
123    )
124
125
126def _album(
127    item_id: str,
128    provider: str,
129    *,
130    name: str = "Album X",
131    version: str = "",
132    artist_names: Sequence[str] = ("Main Artist",),
133    mappings: Sequence[ProviderMapping] | None = None,
134) -> Album:
135    """Build an album for the reference-album leg, with its artists credited by name."""
136    if mappings is None:
137        mappings = [
138            ProviderMapping(item_id=item_id, provider_domain=provider, provider_instance=provider)
139        ]
140    return Album(
141        item_id=item_id,
142        provider=provider,
143        name=name,
144        version=version,
145        artists=UniqueList([_credit(artist_name, provider) for artist_name in artist_names]),
146        provider_mappings=set(mappings),
147    )
148
149
150def _provider() -> Mock:
151    """Return a mock streaming MusicProvider for matching."""
152    provider = Mock()
153    provider.name = "Spotify"
154    provider.instance_id = "spotify_1"
155    provider.domain = "spotify"
156    return provider
157
158
159@dataclass
160class _Harness:
161    """A controller under test together with its mocked IO boundaries."""
162
163    ctrl: ArtistsController
164    track_search: AsyncMock
165    album_search: AsyncMock
166    get_provider_item: AsyncMock
167    provider: Mock
168
169    async def match(self, db_artist: Artist, *, strict: bool = True) -> list[ProviderMapping]:
170        """Match against the single (streaming) provider the harness owns."""
171        return await self.ctrl.match_provider(db_artist, self.provider, strict)
172
173
174@contextmanager
175def _harness(
176    *,
177    ref_tracks: Sequence[Track] = (),
178    track_results: Sequence[Track] = (),
179    ref_albums: Sequence[Album] = (),
180    album_results: Sequence[Album] = (),
181    provider_artists: Sequence[Artist] = (),
182) -> Iterator[_Harness]:
183    """
184    Yield an ArtistsController with every IO boundary mocked.
185
186    :param ref_tracks: Reference tracks of the library artist.
187    :param track_results: Track search results returned by the matched provider.
188    :param ref_albums: Reference albums of the library artist.
189    :param album_results: Album search results returned by the matched provider.
190    :param provider_artists: Full provider artists, resolved by their item id.
191    """
192    full_artists = {artist.item_id: artist for artist in provider_artists}
193
194    async def _artist_tracks(item_id: str, _provider: str) -> list[Track]:
195        return list(ref_tracks) if item_id == LIBRARY_ITEM_ID else []
196
197    async def _artist_albums(item_id: str, _provider: str) -> list[Album]:
198        return list(ref_albums) if item_id == LIBRARY_ITEM_ID else []
199
200    track_search = AsyncMock(return_value=list(track_results))
201    album_search = AsyncMock(return_value=list(album_results))
202    mass = Mock()
203    mass.music.artists.tracks = AsyncMock(side_effect=_artist_tracks)
204    mass.music.artists.albums = AsyncMock(side_effect=_artist_albums)
205    mass.music.tracks.search = track_search
206    mass.music.albums.search = album_search
207    ctrl = ArtistsController.__new__(ArtistsController)
208    ctrl.logger = logging.getLogger("test.artists.match")
209    ctrl.mass = mass
210
211    async def _full_artist(item_id: str, _provider: str, **_kwargs: object) -> Artist:
212        if item_id not in full_artists:
213            raise MediaNotFoundError(item_id)
214        return full_artists[item_id]
215
216    get_provider_item = AsyncMock(side_effect=_full_artist)
217    with patch.multiple(ctrl, get_provider_item=get_provider_item):
218        yield _Harness(ctrl, track_search, album_search, get_provider_item, _provider())
219
220
221# ---------------------------------------------------------------------------
222# reference-track leg
223# ---------------------------------------------------------------------------
224
225
226async def test_corroborating_reference_track_matches() -> None:
227    """A search result the reference track corroborates yields the candidate's mappings."""
228    with _harness(
229        ref_tracks=[_track("lib-track", "library", mappings=(BASE_MAPPING,))],
230        track_results=[_track("s1", "spotify_1")],
231        provider_artists=[_provider_artist()],
232    ) as harness:
233        matches = await harness.match(_library_artist())
234
235    assert matches == [CANDIDATE_MAPPING]
236
237
238async def test_same_title_on_another_recording_does_not_match() -> None:
239    """A result that only shares the track title is not corroboration."""
240    with _harness(
241        ref_tracks=[_track("lib-track", "library", mappings=(BASE_MAPPING,))],
242        track_results=[_track("s1", "spotify_1", album_name="Other Album", duration=260)],
243        provider_artists=[_provider_artist()],
244    ) as harness:
245        matches = await harness.match(_library_artist())
246
247    assert matches == []
248    harness.get_provider_item.assert_not_awaited()
249
250
251async def test_featuring_artist_difference_does_not_block_match() -> None:
252    """A result crediting only the main artist still corroborates a featured-artist track."""
253    with _harness(
254        ref_tracks=[
255            _track(
256                "lib-track",
257                "library",
258                artist_names=("Main Artist", "Feat Artist"),
259                mappings=(BASE_MAPPING,),
260            )
261        ],
262        track_results=[_track("s1", "spotify_1")],
263        provider_artists=[_provider_artist()],
264    ) as harness:
265        matches = await harness.match(_library_artist())
266
267    assert matches == [CANDIDATE_MAPPING]
268
269
270async def test_accent_drift_on_artist_name_matches() -> None:
271    """An accented library artist matches its unaccented spelling on the provider."""
272    with _harness(
273        ref_tracks=[
274            _track("lib-track", "library", artist_names=("Sigur Rós",), mappings=(BASE_MAPPING,))
275        ],
276        track_results=[_track("s1", "spotify_1", artist_names=("Sigur Ros",))],
277        provider_artists=[_provider_artist(name="Sigur Ros")],
278    ) as harness:
279        matches = await harness.match(_library_artist(name="Sigur Rós"))
280
281    assert matches == [CANDIDATE_MAPPING]
282
283
284# ---------------------------------------------------------------------------
285# full-artist confirmation
286# ---------------------------------------------------------------------------
287
288
289async def test_conflicting_musicbrainz_id_on_full_artist_rejects() -> None:
290    """A same-named candidate whose full artist is a different MusicBrainz artist is rejected."""
291    with _harness(
292        ref_tracks=[_track("lib-track", "library", mappings=(BASE_MAPPING,))],
293        track_results=[_track("s1", "spotify_1")],
294        provider_artists=[
295            _provider_artist(external_ids={(ExternalID.MB_ARTIST, OTHER_MB_ARTIST_ID)})
296        ],
297    ) as harness:
298        matches = await harness.match(
299            _library_artist(external_ids={(ExternalID.MB_ARTIST, MB_ARTIST_ID)})
300        )
301
302    assert matches == []
303    # the credit itself carries no external ids, so only the full artist can reject it
304    harness.get_provider_item.assert_awaited_once()
305
306
307async def test_conflicting_artist_type_on_full_artist_rejects() -> None:
308    """A same-named candidate whose full artist is a different artist type is rejected."""
309    with _harness(
310        ref_tracks=[_track("lib-track", "library", mappings=(BASE_MAPPING,))],
311        track_results=[_track("s1", "spotify_1")],
312        provider_artists=[_provider_artist(artist_type=ArtistType.SINGER)],
313    ) as harness:
314        matches = await harness.match(_library_artist(artist_type=ArtistType.AUTHOR))
315
316    assert matches == []
317    harness.get_provider_item.assert_awaited_once()
318
319
320# ---------------------------------------------------------------------------
321# reference-album leg
322# ---------------------------------------------------------------------------
323
324
325async def test_reference_album_edition_difference_confirms_artist() -> None:
326    """An edition difference between the reference album and the result still confirms."""
327    with _harness(
328        ref_albums=[_album("lib-album", "library", mappings=(BASE_MAPPING,))],
329        album_results=[_album("s1", "spotify_1", version="Deluxe Edition")],
330        provider_artists=[_provider_artist()],
331    ) as harness:
332        matches = await harness.match(_library_artist())
333
334    assert matches == [CANDIDATE_MAPPING]
335
336
337async def test_reference_album_packaging_editions_confirm_artist() -> None:
338    """Two non-overlapping packaging editions are still the same record by the same artist."""
339    with _harness(
340        ref_albums=[
341            _album("lib-album", "library", version="2009 Remaster", mappings=(BASE_MAPPING,))
342        ],
343        album_results=[_album("s1", "spotify_1", version="Deluxe Edition")],
344        provider_artists=[_provider_artist()],
345    ) as harness:
346        matches = await harness.match(_library_artist())
347
348    assert matches == [CANDIDATE_MAPPING]
349
350
351async def test_reference_album_retail_suffix_confirms_artist() -> None:
352    """An Apple-style retail suffix on the provider title does not block the match."""
353    with _harness(
354        ref_albums=[_album("lib-album", "library", mappings=(BASE_MAPPING,))],
355        album_results=[_album("s1", "spotify_1", name="Album X - EP")],
356        provider_artists=[_provider_artist()],
357    ) as harness:
358        matches = await harness.match(_library_artist())
359
360    assert matches == [CANDIDATE_MAPPING]
361
362
363async def test_reference_album_different_title_does_not_match() -> None:
364    """A search result for another album cannot confirm the artist."""
365    with _harness(
366        ref_albums=[_album("lib-album", "library", mappings=(BASE_MAPPING,))],
367        album_results=[_album("s1", "spotify_1", name="Another Album")],
368        provider_artists=[_provider_artist()],
369    ) as harness:
370        matches = await harness.match(_library_artist())
371
372    assert matches == []
373    harness.get_provider_item.assert_not_awaited()
374
375
376async def test_album_credit_beyond_the_first_confirms_artist() -> None:
377    """A collaboration album crediting the artist second still confirms it."""
378    with _harness(
379        ref_albums=[_album("lib-album", "library", mappings=(BASE_MAPPING,))],
380        album_results=[_album("s1", "spotify_1", artist_names=("Other Artist", "Main Artist"))],
381        provider_artists=[_provider_artist()],
382    ) as harness:
383        matches = await harness.match(_library_artist())
384
385    assert matches == [CANDIDATE_MAPPING]
386
387
388async def test_unresolvable_credit_does_not_match() -> None:
389    """A credit the provider cannot resolve to a full artist confirms nothing."""
390    with _harness(
391        ref_tracks=[_track("lib-track", "library", mappings=(BASE_MAPPING,))],
392        track_results=[_track("s1", "spotify_1")],
393        provider_artists=[],
394    ) as harness:
395        matches = await harness.match(_library_artist())
396
397    assert matches == []
398
399
400async def test_credit_owned_by_another_library_artist_does_not_match() -> None:
401    """A credit that resolves to a library item belongs to another artist, so it cannot confirm."""
402    other_library_artist = Artist(
403        item_id="lib2",
404        provider="library",
405        name="Main Artist",
406        provider_mappings={BASE_MAPPING, CANDIDATE_MAPPING},
407    )
408    with _harness(
409        ref_tracks=[_track("lib-track", "library", mappings=(BASE_MAPPING,))],
410        track_results=[_track("s1", "spotify_1")],
411    ) as harness:
412        harness.get_provider_item.side_effect = None
413        harness.get_provider_item.return_value = other_library_artist
414        matches = await harness.match(_library_artist())
415
416    assert matches == []
417