music-assistant-server

1"""Tests for the MusicBrainz provider."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, patch
7
8from music_assistant_models.errors import RateLimited
9
10from music_assistant.constants import VARIOUS_ARTISTS_MBID
11from music_assistant.providers.musicbrainz.provider import MusicbrainzProvider
12
13# ---------------------------------------------------------------------------
14# helpers
15# ---------------------------------------------------------------------------
16
17
18def _provider(
19    response: Any, release_group_response: Any = None
20) -> tuple[MusicbrainzProvider, AsyncMock]:
21    """
22    Return a MusicbrainzProvider whose API client answers with the given response.
23
24    :param response: Answer to every request but the release group lookup.
25    :param release_group_response: Answer to the release group lookup.
26    """
27    with patch.object(MusicbrainzProvider, "__init__", lambda *_a, **_kw: None):
28        provider = MusicbrainzProvider.__new__(MusicbrainzProvider)
29
30    async def _answer(endpoint: str, **_kwargs: Any) -> Any:
31        return release_group_response if endpoint == "release-group" else response
32
33    get_data = AsyncMock(side_effect=_answer)
34    api_client = MagicMock()
35    api_client.get_data = get_data
36    provider._api_client = api_client
37    return provider, get_data
38
39
40def _recordings(*first_release_dates: str | None) -> dict[str, Any]:
41    """Return an isrc lookup response with a recording per given release date."""
42    return {
43        "isrc": "GBAYE8600477",
44        "recordings": [
45            {"id": f"stub-{i}", "title": "stub"}
46            if date is None
47            else {"id": f"stub-{i}", "title": "stub", "first-release-date": date}
48            for i, date in enumerate(first_release_dates)
49        ],
50    }
51
52
53# ---------------------------------------------------------------------------
54# get_release_year_by_isrc
55# ---------------------------------------------------------------------------
56
57
58async def test_release_year_parses_all_date_precisions() -> None:
59    """Accept a year, year-month or full date as the first release date."""
60    for release_date in ("1986", "1986-06", "1986-06-23"):
61        provider, _ = _provider(_recordings(release_date))
62        assert await provider.get_release_year_by_isrc("GBAYE8600477") == 1986
63
64
65async def test_release_year_uses_bare_isrc_lookup() -> None:
66    """Look the recording up on the isrc resource without inc parameters."""
67    provider, get_data = _provider(_recordings("1986"))
68
69    await provider.get_release_year_by_isrc("GB-AYE-86-00477")
70
71    get_data.assert_awaited_once_with("isrc/GBAYE8600477")
72
73
74async def test_release_year_returns_earliest_of_multiple_recordings() -> None:
75    """Date the song by the oldest recording the ISRC covers."""
76    provider, _ = _provider(_recordings("2009-05-01", "1986-06", "1994"))
77    assert await provider.get_release_year_by_isrc("GBAYE8600477") == 1986
78
79
80async def test_release_year_is_none_without_a_usable_date() -> None:
81    """Return None when MusicBrainz has no parseable first release date."""
82    for response in (
83        None,
84        {"isrc": "GBAYE8600477"},
85        {"isrc": "GBAYE8600477", "recordings": []},
86        _recordings(None),
87        _recordings("????-06"),
88    ):
89        provider, _ = _provider(response)
90        assert await provider.get_release_year_by_isrc("GBAYE8600477") is None
91
92
93async def test_release_year_rejects_a_malformed_isrc() -> None:
94    """Never put an ISRC that cannot be part of a URL path in the request."""
95    provider, get_data = _provider(_recordings("1986"))
96
97    assert await provider.get_release_year_by_isrc("../artist/1") is None
98    get_data.assert_not_awaited()
99
100
101# ---------------------------------------------------------------------------
102# get_recordings_by_isrc
103# ---------------------------------------------------------------------------
104
105
106_YELLOW_SUBMARINE = {
107    "isrc": "GBAYE0601498",
108    "recordings": [
109        {
110            "id": "b2181aae-5cba-496c-bb0c-b4cc0109ebf8",
111            "title": "Yellow Submarine",
112            "length": 160000,
113            "first-release-date": "1966-08-05",
114            "disambiguation": "original stereo studio mix",
115            "video": False,
116        }
117    ],
118}
119
120
121async def test_recordings_by_isrc_parses_a_realistic_payload() -> None:
122    """Parse id, title and first-release-date from a real isrc lookup response."""
123    provider, _ = _provider(_YELLOW_SUBMARINE)
124
125    recordings = await provider.get_recordings_by_isrc("GBAYE0601498")
126
127    assert len(recordings) == 1
128    recording = recordings[0]
129    assert recording.id == "b2181aae-5cba-496c-bb0c-b4cc0109ebf8"
130    assert recording.title == "Yellow Submarine"
131    assert recording.first_release_date == "1966-08-05"
132
133
134async def test_recordings_by_isrc_returns_all_recordings() -> None:
135    """Return every recording an ISRC covers, not just the first."""
136    provider, _ = _provider(_recordings("2009-05-01", "1986-06", "1994"))
137
138    recordings = await provider.get_recordings_by_isrc("GBAYE8600477")
139
140    assert len(recordings) == 3
141    assert [r.first_release_date for r in recordings] == ["2009-05-01", "1986-06", "1994"]
142
143
144async def test_recordings_by_isrc_skips_a_malformed_entry() -> None:
145    """Skip a recording missing a required field while keeping its valid siblings."""
146    response = {
147        "isrc": "GBAYE8600477",
148        "recordings": [
149            {"title": "no id here"},
150            {"id": "good-1", "title": "stub", "first-release-date": "1986"},
151        ],
152    }
153    provider, _ = _provider(response)
154
155    recordings = await provider.get_recordings_by_isrc("GBAYE8600477")
156
157    assert len(recordings) == 1
158    assert recordings[0].id == "good-1"
159
160
161async def test_recordings_by_isrc_is_empty_without_usable_data() -> None:
162    """Return an empty list for every shape of "MusicBrainz has nothing" response."""
163    for response in (
164        None,
165        {"isrc": "GBAYE8600477"},
166        {"isrc": "GBAYE8600477", "recordings": []},
167    ):
168        provider, _ = _provider(response)
169        assert await provider.get_recordings_by_isrc("GBAYE8600477") == []
170
171
172async def test_recordings_by_isrc_rejects_a_malformed_isrc() -> None:
173    """Never put an ISRC that cannot be part of a URL path in the request."""
174    provider, get_data = _provider(_YELLOW_SUBMARINE)
175
176    assert await provider.get_recordings_by_isrc("../artist/1") == []
177    get_data.assert_not_awaited()
178
179
180# ---------------------------------------------------------------------------
181# get_release_year_by_track_name
182# ---------------------------------------------------------------------------
183
184
185def _credit(name: str, artist_id: str) -> dict[str, Any]:
186    """Return one artist credit of a release."""
187    return {"name": name, "artist": {"id": artist_id, "name": name, "sort-name": name}}
188
189
190def _release(
191    date: str,
192    *,
193    title: str = "A Night at the Opera",
194    primary_type: str = "Album",
195    secondary_types: list[str] | None = None,
196    status: str = "Official",
197    credit: dict[str, Any] | None = None,
198) -> dict[str, Any]:
199    """Return one release of a searched recording."""
200    release: dict[str, Any] = {
201        "id": f"release-{date}",
202        "title": title,
203        "date": date,
204        "status": status,
205        "release-group": {
206            "id": f"rg-{title}-{primary_type}",
207            "title": title,
208            "primary-type": primary_type,
209        },
210    }
211    if secondary_types:
212        release["release-group"]["secondary-types"] = secondary_types
213    if credit is not None:
214        release["artist-credit"] = [credit]
215    return release
216
217
218def _search_result(*recordings: dict[str, Any]) -> dict[str, Any]:
219    """Return a recording search response holding the given recordings."""
220    return {"count": len(recordings), "recordings": list(recordings)}
221
222
223def _recording(
224    *releases: dict[str, Any],
225    title: str = "Bohemian Rhapsody",
226    artist: str = "Queen",
227    artist_id: str = "artist-1",
228    first_release: str | None = None,
229) -> dict[str, Any]:
230    """Return one searched recording credited to the given artist."""
231    recording: dict[str, Any] = {
232        "id": f"recording-{title}-{releases[0]['date'] if releases else 'none'}",
233        "title": title,
234        "artist-credit": [{"artist": {"id": artist_id, "name": artist, "sort-name": artist}}],
235        "releases": list(releases),
236    }
237    if first_release is not None:
238        recording["first-release-date"] = first_release
239    return recording
240
241
242async def test_release_year_by_track_name_returns_the_earliest_studio_release() -> None:
243    """Date a song by the oldest studio album any matching recording appeared on."""
244    provider, get_data = _provider(
245        _search_result(
246            _recording(_release("2011-05-16", title="The Platinum Collection")),
247            _recording(_release("1992-08-25", title="Classic Queen")),
248            _recording(_release("1975-11-21")),
249        )
250    )
251
252    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
253    assert get_data.await_args_list[0].args == ("recording",)
254    assert get_data.await_args_list[0].kwargs == {
255        "query": '"Bohemian Rhapsody" AND artist:"Queen"',
256        "limit": "100",
257    }
258
259
260async def test_release_year_by_track_name_ignores_untrustworthy_releases() -> None:
261    """Never date a song by a compilation, a live album, a bootleg or an unrelated single."""
262    provider, _ = _provider(
263        _search_result(
264            _recording(
265                # every untrusted release predates the studio album, so each filter has to
266                # hold on its own for the studio year to win
267                _release("1968-10-26", title="Greatest Hits", secondary_types=["Compilation"]),
268                _release("1969-06-22", title="Live Killers", secondary_types=["Live"]),
269                _release("1970-01-01", title="Some Other Song", primary_type="Single"),
270                _release("1971-03-01", title="Bootleg Tape", status="Bootleg"),
271                _release("1972-05-05", title="A Tribute", primary_type="Other"),
272                _release("1975-11-21"),
273            )
274        )
275    )
276
277    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
278
279
280async def test_release_year_by_track_name_ignores_a_various_artists_release() -> None:
281    """Never date a song by a hits compilation that carries no Compilation type."""
282    provider, _ = _provider(
283        _search_result(
284            _recording(
285                # the compilation predates the studio album, so the credit filter has to
286                # hold on its own for the studio year to win
287                _release(
288                    "1968-10-26",
289                    title="Hits of the 60s",
290                    credit=_credit("Various Artists", VARIOUS_ARTISTS_MBID),
291                ),
292                _release("1975-11-21", credit=_credit("Queen", "artist-1")),
293            )
294        )
295    )
296
297    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
298
299
300async def test_release_year_by_track_name_identifies_various_artists_by_id() -> None:
301    """
302    Recognise the Various Artists entity by its id rather than by its name.
303
304    MusicBrainz localizes the name it credits that entity under, and unrelated artists
305    are named after it.
306    """
307    provider, _ = _provider(
308        _search_result(
309            _recording(
310                _release(
311                    "1968-10-26",
312                    title="Artisti Vari Compilation",
313                    credit=_credit("Artisti Vari", VARIOUS_ARTISTS_MBID),
314                ),
315                _release(
316                    "1975-11-21",
317                    credit=_credit("Various Artist", "artist-named-like-various"),
318                ),
319            )
320        )
321    )
322
323    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
324
325
326async def test_release_group_by_track_name_drops_a_various_artists_release() -> None:
327    """Offer no artwork candidate when a song is only listed on a hits compilation."""
328    provider, _ = _provider(
329        _search_result(
330            _recording(
331                _release(
332                    "1981-10-26",
333                    title="Hits of the 80s",
334                    credit=_credit("Various Artists", VARIOUS_ARTISTS_MBID),
335                )
336            )
337        )
338    )
339
340    result = await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
341
342    assert result is not None
343    artist, release_groups = result
344    assert artist.name == "Queen"
345    assert release_groups == []
346
347
348async def test_release_year_by_track_name_ignores_an_undated_release_group() -> None:
349    """Date a song by the oldest release group that has a date, not by an undated one."""
350    provider, _ = _provider(
351        _search_result(
352            _recording(_release("", title="Unknown Pressing")),
353            _recording(_release("1975-11-21")),
354        )
355    )
356
357    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
358
359
360async def test_release_year_by_track_name_accepts_a_single_named_after_the_song() -> None:
361    """Date a song by its own single when no studio album carries it."""
362    provider, _ = _provider(
363        _search_result(
364            _recording(_release("1975-10-31", title="Bohemian Rhapsody", primary_type="Single"))
365        )
366    )
367
368    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
369
370
371async def test_release_year_by_track_name_is_none_without_a_confident_match() -> None:
372    """Return no year at all rather than guessing from a name that does not match."""
373    for response in (
374        None,
375        {"count": 0, "recordings": []},
376        _search_result(_recording(_release("1975-11-21"), artist="Not Queen")),
377        _search_result(_recording(_release("1975-11-21"), title="Another Song")),
378        _search_result(_recording()),
379        _search_result(_recording(_release(""))),
380    ):
381        provider, _ = _provider(response)
382        assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") is None
383
384
385def _release_groups(*groups: tuple[str, str]) -> dict[str, Any]:
386    """
387    Return a release group search response holding the given release groups.
388
389    :param groups: Release groups as (id, first release date) pairs.
390    """
391    return {
392        "count": len(groups),
393        "release-groups": [
394            {"id": group_id, "title": group_id, "first-release-date": date}
395            for group_id, date in groups
396        ],
397    }
398
399
400async def test_release_year_by_track_name_prefers_the_release_group_first_release() -> None:
401    """Date a much reissued song by its album's first release, not by the reissue found."""
402    provider, _ = _provider(
403        _search_result(_recording(_release("2021-11-12"))),
404        _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
405    )
406
407    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
408
409
410async def test_release_year_by_track_name_keeps_a_close_release_date() -> None:
411    """Keep the found release when the album barely predates it, as a single ahead of it would."""
412    provider, _ = _provider(
413        _search_result(_recording(_release("1975-11-21"))),
414        _release_groups(("rg-A Night at the Opera-Album", "1974-10-31")),
415    )
416
417    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
418
419
420async def test_release_year_by_track_name_corrects_only_beyond_the_threshold() -> None:
421    """Correct the year only once the album predates the found release by enough years."""
422    for first_release_year, expected in ((1970, 1975), (1969, 1969)):
423        provider, _ = _provider(
424            _search_result(_recording(_release("1975-11-21"))),
425            _release_groups(("rg-A Night at the Opera-Album", f"{first_release_year}-10-31")),
426        )
427
428        assert (
429            await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == expected
430        )
431
432
433async def test_release_year_by_track_name_keeps_the_found_release_when_the_lookup_fails() -> None:
434    """Never lose the year the search already supplied when the release group lookup fails."""
435    search_result = _search_result(_recording(_release("1975-11-21")))
436    provider, get_data = _provider(search_result)
437
438    async def _answer(endpoint: str, **_kwargs: Any) -> Any:
439        if endpoint == "release-group":
440            raise RateLimited("rate limited")
441        return search_result
442
443    get_data.side_effect = _answer
444
445    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
446
447
448async def test_release_year_by_track_name_looks_up_every_release_group_at_once() -> None:
449    """Resolve all release groups of a song with a single, escaped, request."""
450    provider, get_data = _provider(
451        _search_result(
452            _recording(_release("2011-05-16", title="The Platinum Collection")),
453            _recording(_release("1975-11-21")),
454        ),
455        _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
456    )
457
458    await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody")
459
460    # one release group lookup, however many groups the search turned up
461    assert [call.args for call in get_data.await_args_list] == [("recording",), ("release-group",)]
462    assert get_data.await_args_list[1].kwargs["query"] == (
463        r"rgid:(rg\-A Night at the Opera\-Album OR rg\-The Platinum Collection\-Album)"
464    )
465
466
467async def test_release_year_by_track_name_falls_back_to_the_found_release() -> None:
468    """Keep the found release when MusicBrainz does not know when the album first came out."""
469    for release_group_response in (None, _release_groups(), {"release-groups": [{"id": "rg-x"}]}):
470        provider, _ = _provider(
471            _search_result(_recording(_release("1975-11-21"))), release_group_response
472        )
473
474        assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
475
476
477async def test_release_year_by_track_name_dates_an_undated_release_group() -> None:
478    """Date a song whose found releases carry no date at all by its album's first release."""
479    provider, _ = _provider(
480        _search_result(_recording(_release(""))),
481        _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
482    )
483
484    assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
485
486
487async def test_release_group_by_track_name_costs_a_single_request() -> None:
488    """Never spend the release group lookup on callers that only want the release groups."""
489    provider, get_data = _provider(
490        _search_result(_recording(_release("1975-11-21"))),
491        _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
492    )
493
494    await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
495
496    assert [call.args for call in get_data.await_args_list] == [("recording",)]
497
498
499async def test_release_group_by_track_name_returns_the_artist_and_oldest_groups_first() -> None:
500    """Hand out the artist of the oldest recording, and their release groups oldest first."""
501    provider, _ = _provider(
502        _search_result(
503            _recording(
504                _release("2011-05-16", title="The Platinum Collection"),
505                first_release="2011-05-16",
506                artist_id="artist-reissue",
507            ),
508            _recording(
509                _release("1975-11-21"),
510                first_release="1975-11-21",
511                artist_id="artist-original",
512            ),
513        )
514    )
515
516    result = await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
517
518    assert result is not None
519    artist, release_groups = result
520    # MusicBrainz can hold several artist entries under one name, and the oldest recording
521    # is the one that identifies the original
522    assert artist.id == "artist-original"
523    assert [group.title for group in release_groups] == [
524        "A Night at the Opera",
525        "The Platinum Collection",
526    ]
527
528
529async def test_release_group_by_track_name_returns_the_artist_without_release_groups() -> None:
530    """Still identify the artist when no matched recording carries a usable release group."""
531    provider, _ = _provider(
532        _search_result(_recording(_release("1981-10-26", secondary_types=["Compilation"])))
533    )
534
535    result = await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
536
537    assert result is not None
538    artist, release_groups = result
539    assert artist.name == "Queen"
540    assert release_groups == []
541
542
543async def test_release_year_by_track_name_escapes_lucene_specials() -> None:
544    """Escape characters that would otherwise change the meaning of the search query."""
545    provider, get_data = _provider(_search_result())
546
547    await provider.get_release_year_by_track_name("AC/DC", "T.N.T. (live!)")
548
549    get_data.assert_awaited_once_with(
550        "recording",
551        query='"T.N.T. \\(live\\!\\)" AND artist:"AC\\/DC"',
552        limit="100",
553    )
554