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