/
/
/
1"""Tests for the MusicBrainz provider."""
2
3from __future__ import annotations
4
5from typing import Any
6from unittest.mock import AsyncMock, MagicMock, patch
7
8import pytest
9from music_assistant_models.errors import InvalidDataError, RateLimited
10
11from music_assistant.constants import VARIOUS_ARTISTS_MBID
12from music_assistant.providers.musicbrainz.provider import MusicbrainzProvider
13
14# ---------------------------------------------------------------------------
15# helpers
16# ---------------------------------------------------------------------------
17
18
19def _provider(
20 response: Any, release_group_response: Any = None
21) -> tuple[MusicbrainzProvider, AsyncMock]:
22 """
23 Return a MusicbrainzProvider whose API client answers with the given response.
24
25 :param response: Answer to every request but the release group lookup.
26 :param release_group_response: Answer to the release group lookup.
27 """
28 with patch.object(MusicbrainzProvider, "__init__", lambda *_a, **_kw: None):
29 provider = MusicbrainzProvider.__new__(MusicbrainzProvider)
30
31 async def _answer(endpoint: str, **_kwargs: Any) -> Any:
32 return release_group_response if endpoint == "release-group" else response
33
34 get_data = AsyncMock(side_effect=_answer)
35 api_client = MagicMock()
36 api_client.get_data = get_data
37 provider._api_client = api_client
38 return provider, get_data
39
40
41def _recordings(*first_release_dates: str | None) -> dict[str, Any]:
42 """Return an isrc lookup response with a recording per given release date."""
43 return {
44 "isrc": "GBAYE8600477",
45 "recordings": [
46 {"id": f"stub-{i}", "title": "stub"}
47 if date is None
48 else {"id": f"stub-{i}", "title": "stub", "first-release-date": date}
49 for i, date in enumerate(first_release_dates)
50 ],
51 }
52
53
54# ---------------------------------------------------------------------------
55# get_release_year_by_isrc
56# ---------------------------------------------------------------------------
57
58
59async def test_release_year_parses_all_date_precisions() -> None:
60 """Accept a year, year-month or full date as the first release date."""
61 for release_date in ("1986", "1986-06", "1986-06-23"):
62 provider, _ = _provider(_recordings(release_date))
63 assert await provider.get_release_year_by_isrc("GBAYE8600477") == 1986
64
65
66async def test_release_year_uses_bare_isrc_lookup() -> None:
67 """Look the recording up on the isrc resource without inc parameters."""
68 provider, get_data = _provider(_recordings("1986"))
69
70 await provider.get_release_year_by_isrc("GB-AYE-86-00477")
71
72 get_data.assert_awaited_once_with("isrc/GBAYE8600477")
73
74
75async def test_release_year_returns_earliest_of_multiple_recordings() -> None:
76 """Date the song by the oldest recording the ISRC covers."""
77 provider, _ = _provider(_recordings("2009-05-01", "1986-06", "1994"))
78 assert await provider.get_release_year_by_isrc("GBAYE8600477") == 1986
79
80
81async def test_release_year_is_none_without_a_usable_date() -> None:
82 """Return None when MusicBrainz has no parseable first release date."""
83 for response in (
84 None,
85 {"isrc": "GBAYE8600477"},
86 {"isrc": "GBAYE8600477", "recordings": []},
87 _recordings(None),
88 _recordings("????-06"),
89 ):
90 provider, _ = _provider(response)
91 assert await provider.get_release_year_by_isrc("GBAYE8600477") is None
92
93
94async def test_release_year_rejects_a_malformed_isrc() -> None:
95 """Never put an ISRC that cannot be part of a URL path in the request."""
96 provider, get_data = _provider(_recordings("1986"))
97
98 assert await provider.get_release_year_by_isrc("../artist/1") is None
99 get_data.assert_not_awaited()
100
101
102# ---------------------------------------------------------------------------
103# get_recordings_by_isrc
104# ---------------------------------------------------------------------------
105
106
107_YELLOW_SUBMARINE = {
108 "isrc": "GBAYE0601498",
109 "recordings": [
110 {
111 "id": "b2181aae-5cba-496c-bb0c-b4cc0109ebf8",
112 "title": "Yellow Submarine",
113 "length": 160000,
114 "first-release-date": "1966-08-05",
115 "disambiguation": "original stereo studio mix",
116 "video": False,
117 }
118 ],
119}
120
121
122async def test_recordings_by_isrc_parses_a_realistic_payload() -> None:
123 """Parse id, title and first-release-date from a real isrc lookup response."""
124 provider, _ = _provider(_YELLOW_SUBMARINE)
125
126 recordings = await provider.get_recordings_by_isrc("GBAYE0601498")
127
128 assert len(recordings) == 1
129 recording = recordings[0]
130 assert recording.id == "b2181aae-5cba-496c-bb0c-b4cc0109ebf8"
131 assert recording.title == "Yellow Submarine"
132 assert recording.first_release_date == "1966-08-05"
133
134
135async def test_recordings_by_isrc_returns_all_recordings() -> None:
136 """Return every recording an ISRC covers, not just the first."""
137 provider, _ = _provider(_recordings("2009-05-01", "1986-06", "1994"))
138
139 recordings = await provider.get_recordings_by_isrc("GBAYE8600477")
140
141 assert len(recordings) == 3
142 assert [r.first_release_date for r in recordings] == ["2009-05-01", "1986-06", "1994"]
143
144
145async def test_recordings_by_isrc_skips_a_malformed_entry() -> None:
146 """Skip a recording missing a required field while keeping its valid siblings."""
147 response = {
148 "isrc": "GBAYE8600477",
149 "recordings": [
150 {"title": "no id here"},
151 {"id": "good-1", "title": "stub", "first-release-date": "1986"},
152 ],
153 }
154 provider, _ = _provider(response)
155
156 recordings = await provider.get_recordings_by_isrc("GBAYE8600477")
157
158 assert len(recordings) == 1
159 assert recordings[0].id == "good-1"
160
161
162async def test_recordings_by_isrc_is_empty_without_usable_data() -> None:
163 """Return an empty list for every shape of "MusicBrainz has nothing" response."""
164 for response in (
165 None,
166 {"isrc": "GBAYE8600477"},
167 {"isrc": "GBAYE8600477", "recordings": []},
168 ):
169 provider, _ = _provider(response)
170 assert await provider.get_recordings_by_isrc("GBAYE8600477") == []
171
172
173async def test_recordings_by_isrc_rejects_a_malformed_isrc() -> None:
174 """Never put an ISRC that cannot be part of a URL path in the request."""
175 provider, get_data = _provider(_YELLOW_SUBMARINE)
176
177 assert await provider.get_recordings_by_isrc("../artist/1") == []
178 get_data.assert_not_awaited()
179
180
181# ---------------------------------------------------------------------------
182# get_release_year_by_track_name
183# ---------------------------------------------------------------------------
184
185
186def _credit(name: str, artist_id: str) -> dict[str, Any]:
187 """Return one artist credit of a release."""
188 return {"name": name, "artist": {"id": artist_id, "name": name, "sort-name": name}}
189
190
191def _release(
192 date: str,
193 *,
194 title: str = "A Night at the Opera",
195 primary_type: str = "Album",
196 secondary_types: list[str] | None = None,
197 status: str = "Official",
198 credit: dict[str, Any] | None = None,
199) -> dict[str, Any]:
200 """Return one release of a searched recording."""
201 release: dict[str, Any] = {
202 "id": f"release-{date}",
203 "title": title,
204 "date": date,
205 "status": status,
206 "release-group": {
207 "id": f"rg-{title}-{primary_type}",
208 "title": title,
209 "primary-type": primary_type,
210 },
211 }
212 if secondary_types:
213 release["release-group"]["secondary-types"] = secondary_types
214 if credit is not None:
215 release["artist-credit"] = [credit]
216 return release
217
218
219def _search_result(*recordings: dict[str, Any]) -> dict[str, Any]:
220 """Return a recording search response holding the given recordings."""
221 return {"count": len(recordings), "recordings": list(recordings)}
222
223
224def _recording(
225 *releases: dict[str, Any],
226 title: str = "Bohemian Rhapsody",
227 artist: str = "Queen",
228 artist_id: str = "artist-1",
229 first_release: str | None = None,
230) -> dict[str, Any]:
231 """Return one searched recording credited to the given artist."""
232 recording: dict[str, Any] = {
233 "id": f"recording-{title}-{releases[0]['date'] if releases else 'none'}",
234 "title": title,
235 "artist-credit": [{"artist": {"id": artist_id, "name": artist, "sort-name": artist}}],
236 "releases": list(releases),
237 }
238 if first_release is not None:
239 recording["first-release-date"] = first_release
240 return recording
241
242
243async def test_release_year_by_track_name_returns_the_earliest_studio_release() -> None:
244 """Date a song by the oldest studio album any matching recording appeared on."""
245 provider, get_data = _provider(
246 _search_result(
247 _recording(_release("2011-05-16", title="The Platinum Collection")),
248 _recording(_release("1992-08-25", title="Classic Queen")),
249 _recording(_release("1975-11-21")),
250 )
251 )
252
253 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
254 assert get_data.await_args_list[0].args == ("recording",)
255 assert get_data.await_args_list[0].kwargs == {
256 "query": '"Bohemian Rhapsody" AND artist:"Queen"',
257 "limit": "100",
258 }
259
260
261async def test_release_year_by_track_name_ignores_untrustworthy_releases() -> None:
262 """Never date a song by a compilation, a live album, a bootleg or an unrelated single."""
263 provider, _ = _provider(
264 _search_result(
265 _recording(
266 # every untrusted release predates the studio album, so each filter has to
267 # hold on its own for the studio year to win
268 _release("1968-10-26", title="Greatest Hits", secondary_types=["Compilation"]),
269 _release("1969-06-22", title="Live Killers", secondary_types=["Live"]),
270 _release("1970-01-01", title="Some Other Song", primary_type="Single"),
271 _release("1971-03-01", title="Bootleg Tape", status="Bootleg"),
272 _release("1972-05-05", title="A Tribute", primary_type="Other"),
273 _release("1975-11-21"),
274 )
275 )
276 )
277
278 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
279
280
281async def test_release_year_by_track_name_dates_a_song_by_its_soundtrack() -> None:
282 """Date a song written for a film by that film's soundtrack."""
283 provider, _ = _provider(
284 _search_result(
285 _recording(
286 _release(
287 "1994-05-31",
288 title="The Lion King: Original Motion Picture Soundtrack",
289 secondary_types=["Soundtrack"],
290 ),
291 _release("2013-09-13", title="The Diving Board"),
292 title="Circle of Life",
293 artist="Elton John",
294 )
295 )
296 )
297
298 assert await provider.get_release_year_by_track_name("Elton John", "Circle of Life") == 1994
299
300
301async def test_release_year_by_track_name_ignores_a_soundtrack_compilation() -> None:
302 """Never date a song by a film compilation of songs released before it."""
303 provider, _ = _provider(
304 _search_result(
305 _recording(
306 # the compilation predates the studio album, so the secondary type filter
307 # has to hold on its own for the studio year to win
308 _release(
309 "1968-10-26",
310 title="Music From the Motion Picture",
311 secondary_types=["Compilation", "Soundtrack"],
312 ),
313 _release("1975-11-21"),
314 )
315 )
316 )
317
318 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
319
320
321async def test_release_year_by_track_name_ignores_a_various_artists_soundtrack() -> None:
322 """Never date a song by a film compilation credited to Various Artists."""
323 provider, _ = _provider(
324 _search_result(
325 _recording(
326 # most film soundtracks are compilations of several artists, and the credit
327 # filter is what keeps them out now that soundtracks are allowed through
328 _release(
329 "1968-10-26",
330 title="Music From the Motion Picture",
331 secondary_types=["Soundtrack"],
332 credit=_credit("Various Artists", VARIOUS_ARTISTS_MBID),
333 ),
334 _release("1975-11-21", credit=_credit("Queen", "artist-1")),
335 )
336 )
337 )
338
339 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
340
341
342async def test_release_year_by_track_name_ignores_a_various_artists_release() -> None:
343 """Never date a song by a hits compilation that carries no Compilation type."""
344 provider, _ = _provider(
345 _search_result(
346 _recording(
347 # the compilation predates the studio album, so the credit filter has to
348 # hold on its own for the studio year to win
349 _release(
350 "1968-10-26",
351 title="Hits of the 60s",
352 credit=_credit("Various Artists", VARIOUS_ARTISTS_MBID),
353 ),
354 _release("1975-11-21", credit=_credit("Queen", "artist-1")),
355 )
356 )
357 )
358
359 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
360
361
362async def test_release_year_by_track_name_identifies_various_artists_by_id() -> None:
363 """
364 Recognise the Various Artists entity by its id rather than by its name.
365
366 MusicBrainz localizes the name it credits that entity under, and unrelated artists
367 are named after it.
368 """
369 provider, _ = _provider(
370 _search_result(
371 _recording(
372 _release(
373 "1968-10-26",
374 title="Artisti Vari Compilation",
375 credit=_credit("Artisti Vari", VARIOUS_ARTISTS_MBID),
376 ),
377 _release(
378 "1975-11-21",
379 credit=_credit("Various Artist", "artist-named-like-various"),
380 ),
381 )
382 )
383 )
384
385 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
386
387
388async def test_release_group_by_track_name_drops_a_various_artists_release() -> None:
389 """Offer no artwork candidate when a song is only listed on a hits compilation."""
390 provider, _ = _provider(
391 _search_result(
392 _recording(
393 _release(
394 "1981-10-26",
395 title="Hits of the 80s",
396 credit=_credit("Various Artists", VARIOUS_ARTISTS_MBID),
397 )
398 )
399 )
400 )
401
402 result = await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
403
404 assert result is not None
405 artist, release_groups = result
406 assert artist.name == "Queen"
407 assert release_groups == []
408
409
410async def test_release_group_by_track_name_offers_a_soundtrack_as_artwork() -> None:
411 """Offer the soundtrack of a song written for a film as an artwork candidate."""
412 provider, _ = _provider(
413 _search_result(
414 _recording(
415 _release(
416 "1994-05-31",
417 title="The Lion King: Original Motion Picture Soundtrack",
418 secondary_types=["Soundtrack"],
419 ),
420 title="Circle of Life",
421 artist="Elton John",
422 )
423 )
424 )
425
426 result = await provider.get_release_group_by_track_name("Elton John", "Circle of Life")
427
428 assert result is not None
429 _, release_groups = result
430 assert [rg.title for rg in release_groups] == [
431 "The Lion King: Original Motion Picture Soundtrack"
432 ]
433
434
435async def test_release_year_by_track_name_ignores_an_undated_release_group() -> None:
436 """Date a song by the oldest release group that has a date, not by an undated one."""
437 provider, _ = _provider(
438 _search_result(
439 _recording(_release("", title="Unknown Pressing")),
440 _recording(_release("1975-11-21")),
441 )
442 )
443
444 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
445
446
447async def test_release_year_by_track_name_accepts_a_single_named_after_the_song() -> None:
448 """Date a song by its own single when no studio album carries it."""
449 provider, _ = _provider(
450 _search_result(
451 _recording(_release("1975-10-31", title="Bohemian Rhapsody", primary_type="Single"))
452 )
453 )
454
455 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
456
457
458async def test_release_year_by_track_name_is_none_without_a_confident_match() -> None:
459 """Return no year at all rather than guessing from a name that does not match."""
460 for response in (
461 None,
462 {"count": 0, "recordings": []},
463 _search_result(_recording(_release("1975-11-21"), artist="Not Queen")),
464 _search_result(_recording(_release("1975-11-21"), title="Another Song")),
465 _search_result(_recording()),
466 _search_result(_recording(_release(""))),
467 ):
468 provider, _ = _provider(response)
469 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") is None
470
471
472def _release_groups(*groups: tuple[str, str]) -> dict[str, Any]:
473 """
474 Return a release group search response holding the given release groups.
475
476 :param groups: Release groups as (id, first release date) pairs.
477 """
478 return {
479 "count": len(groups),
480 "release-groups": [
481 {"id": group_id, "title": group_id, "first-release-date": date}
482 for group_id, date in groups
483 ],
484 }
485
486
487async def test_release_year_by_track_name_prefers_the_release_group_first_release() -> None:
488 """Date a much reissued song by its album's first release, not by the reissue found."""
489 provider, _ = _provider(
490 _search_result(_recording(_release("2021-11-12"))),
491 _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
492 )
493
494 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
495
496
497async def test_release_year_by_track_name_keeps_a_close_release_date() -> None:
498 """Keep the found release when the album barely predates it, as a single ahead of it would."""
499 provider, _ = _provider(
500 _search_result(_recording(_release("1975-11-21"))),
501 _release_groups(("rg-A Night at the Opera-Album", "1974-10-31")),
502 )
503
504 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
505
506
507async def test_release_year_by_track_name_corrects_only_beyond_the_threshold() -> None:
508 """Correct the year only once the album predates the found release by enough years."""
509 for first_release_year, expected in ((1970, 1975), (1969, 1969)):
510 provider, _ = _provider(
511 _search_result(_recording(_release("1975-11-21"))),
512 _release_groups(("rg-A Night at the Opera-Album", f"{first_release_year}-10-31")),
513 )
514
515 assert (
516 await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == expected
517 )
518
519
520async def test_release_year_by_track_name_keeps_the_found_release_when_the_lookup_fails() -> None:
521 """Never lose the year the search already supplied when the release group lookup fails."""
522 search_result = _search_result(_recording(_release("1975-11-21")))
523 provider, get_data = _provider(search_result)
524
525 async def _answer(endpoint: str, **_kwargs: Any) -> Any:
526 if endpoint == "release-group":
527 raise RateLimited("rate limited")
528 return search_result
529
530 get_data.side_effect = _answer
531
532 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
533
534
535async def test_release_year_by_track_name_looks_up_every_release_group_at_once() -> None:
536 """Resolve all release groups of a song with a single, escaped, request."""
537 provider, get_data = _provider(
538 _search_result(
539 _recording(_release("2011-05-16", title="The Platinum Collection")),
540 _recording(_release("1975-11-21")),
541 ),
542 _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
543 )
544
545 await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody")
546
547 # one release group lookup, however many groups the search turned up
548 assert [call.args for call in get_data.await_args_list] == [("recording",), ("release-group",)]
549 assert get_data.await_args_list[1].kwargs["query"] == (
550 r"rgid:(rg\-A Night at the Opera\-Album OR rg\-The Platinum Collection\-Album)"
551 )
552
553
554async def test_release_year_by_track_name_falls_back_to_the_found_release() -> None:
555 """Keep the found release when MusicBrainz does not know when the album first came out."""
556 for release_group_response in (None, _release_groups(), {"release-groups": [{"id": "rg-x"}]}):
557 provider, _ = _provider(
558 _search_result(_recording(_release("1975-11-21"))), release_group_response
559 )
560
561 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
562
563
564async def test_release_year_by_track_name_dates_an_undated_release_group() -> None:
565 """Date a song whose found releases carry no date at all by its album's first release."""
566 provider, _ = _provider(
567 _search_result(_recording(_release(""))),
568 _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
569 )
570
571 assert await provider.get_release_year_by_track_name("Queen", "Bohemian Rhapsody") == 1975
572
573
574async def test_release_group_by_track_name_costs_a_single_request() -> None:
575 """Never spend the release group lookup on callers that only want the release groups."""
576 provider, get_data = _provider(
577 _search_result(_recording(_release("1975-11-21"))),
578 _release_groups(("rg-A Night at the Opera-Album", "1975-11-21")),
579 )
580
581 await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
582
583 assert [call.args for call in get_data.await_args_list] == [("recording",)]
584
585
586async def test_release_group_by_track_name_returns_the_artist_and_oldest_groups_first() -> None:
587 """Hand out the artist of the oldest recording, and their release groups oldest first."""
588 provider, _ = _provider(
589 _search_result(
590 _recording(
591 _release("2011-05-16", title="The Platinum Collection"),
592 first_release="2011-05-16",
593 artist_id="artist-reissue",
594 ),
595 _recording(
596 _release("1975-11-21"),
597 first_release="1975-11-21",
598 artist_id="artist-original",
599 ),
600 )
601 )
602
603 result = await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
604
605 assert result is not None
606 artist, release_groups = result
607 # MusicBrainz can hold several artist entries under one name, and the oldest recording
608 # is the one that identifies the original
609 assert artist.id == "artist-original"
610 assert [group.title for group in release_groups] == [
611 "A Night at the Opera",
612 "The Platinum Collection",
613 ]
614
615
616async def test_release_group_by_track_name_returns_the_artist_without_release_groups() -> None:
617 """Still identify the artist when no matched recording carries a usable release group."""
618 provider, _ = _provider(
619 _search_result(_recording(_release("1981-10-26", secondary_types=["Compilation"])))
620 )
621
622 result = await provider.get_release_group_by_track_name("Queen", "Bohemian Rhapsody")
623
624 assert result is not None
625 artist, release_groups = result
626 assert artist.name == "Queen"
627 assert release_groups == []
628
629
630async def test_release_year_by_track_name_escapes_lucene_specials() -> None:
631 """Escape characters that would otherwise change the meaning of the search query."""
632 provider, get_data = _provider(_search_result())
633
634 await provider.get_release_year_by_track_name("AC/DC", "T.N.T. (live!)")
635
636 get_data.assert_awaited_once_with(
637 "recording",
638 query='"T.N.T. \\(live\\!\\)" AND artist:"AC\\/DC"',
639 limit="100",
640 )
641
642
643# ---------------------------------------------------------------------------
644# get_releases_by_barcode
645# ---------------------------------------------------------------------------
646
647
648def _barcode_release(release_id: str, release_group_id: str, barcode: str) -> dict[str, Any]:
649 """
650 Return one release stub as a barcode search actually returns it.
651
652 Includes the summary ``media`` object (format/track-count, no tracklist) that the
653 full release model cannot parse, so the slim search model is exercised realistically.
654 """
655 return {
656 "id": release_id,
657 "status-id": "status-id",
658 "count": 1,
659 "title": "( )",
660 "status": "Official",
661 "barcode": barcode,
662 "artist-credit": [_credit("Sigur Rós", "artist-1")],
663 "release-group": {"id": release_group_id, "title": "( )", "primary-type": "Album"},
664 "media": [{"format": "CD", "disc-count": 1, "track-count": 14}],
665 "track-count": 14,
666 }
667
668
669async def test_releases_by_barcode_parses_releases() -> None:
670 """Return every release MusicBrainz has on file for a barcode (summary media and all)."""
671 response = {
672 "count": 2,
673 "releases": [
674 _barcode_release("rel-1", "rg-1", "0888072439412"),
675 _barcode_release("rel-2", "rg-1", "0888072439412"),
676 ],
677 }
678 provider, _ = _provider(response)
679
680 releases = await provider.get_releases_by_barcode("888072439412")
681
682 assert [release.id for release in releases] == ["rel-1", "rel-2"]
683 assert {release.release_group.id for release in releases} == {"rg-1"}
684
685
686async def test_releases_by_barcode_queries_every_compatible_form() -> None:
687 """Query the UPC-12 and its zero-padded EAN-13/GTIN forms in a single request."""
688 provider, get_data = _provider({"releases": []})
689
690 await provider.get_releases_by_barcode("888072439412")
691
692 get_data.assert_awaited_once()
693 call = get_data.await_args
694 assert call is not None
695 assert call.args == ("release",)
696 assert call.kwargs["limit"] == "100"
697 query = call.kwargs["query"]
698 assert "barcode:888072439412" in query
699 assert "barcode:0888072439412" in query
700
701
702async def test_releases_by_barcode_skips_an_invalid_barcode() -> None:
703 """A structurally invalid barcode is treated as absent, without any request."""
704 provider, get_data = _provider({"releases": []})
705
706 assert await provider.get_releases_by_barcode("not-a-barcode") == []
707 get_data.assert_not_awaited()
708
709
710async def test_releases_by_barcode_is_empty_when_not_found() -> None:
711 """An unknown barcode yields an empty list rather than an error."""
712 provider, _ = _provider(None)
713
714 assert await provider.get_releases_by_barcode("888072439412") == []
715
716
717async def test_releases_by_barcode_abstains_on_malformed_entry() -> None:
718 """One unparsable release makes the whole lookup abstain rather than look complete."""
719 response = {
720 "releases": [
721 {"id": "broken"},
722 _barcode_release("rel-2", "rg-2", "0888072439412"),
723 ]
724 }
725 provider, _ = _provider(response)
726
727 with pytest.raises(InvalidDataError):
728 await provider.get_releases_by_barcode("888072439412")
729
730
731async def test_releases_by_barcode_abstains_on_truncated_result() -> None:
732 """A truncated page abstains rather than treating a partial set as complete."""
733 response = {
734 "count": 5,
735 "releases": [_barcode_release("rel-1", "rg-1", "0888072439412")],
736 }
737 provider, _ = _provider(response)
738
739 with pytest.raises(InvalidDataError):
740 await provider.get_releases_by_barcode("888072439412")
741