import musicbrainzngs import pytest from lyra_worker._musicbrainz import _best_release_group, _with_retry # Offline unit tests for the pure release-group selection logic. The live resolve() # path (network) is covered by test_musicbrainz_live.py. def _rg(title, primary_type, artist=""): g = {"id": title.lower(), "title": title, "primary-type": primary_type} if artist: g["artist-credit"] = [{"artist": {"name": artist, "id": artist.lower()}}] return g def test_prefers_album_over_same_titled_single(): # Real-world "Off the Wall" bug: MusicBrainz returns the Single release-group # first, tied on title with the Album. We must pick the Album so a 10-track # download isn't tagged from the 2-track single's tracklist. groups = [ _rg("Off the Wall", "Single", "Michael Jackson"), _rg("Off the Wall", "Album", "Michael Jackson"), ] assert _best_release_group("Michael Jackson", "Off the Wall", groups)["primary-type"] == "Album" def test_album_preference_is_order_independent(): groups = [ _rg("Off the Wall", "Album", "Michael Jackson"), _rg("Off the Wall", "Single", "Michael Jackson"), ] assert _best_release_group("Michael Jackson", "Off the Wall", groups)["primary-type"] == "Album" def test_closer_title_still_wins_over_album_type(): # Title similarity dominates the album-type preference (for a single artist). groups = [ _rg("Continuum", "Single", "John Mayer"), _rg("Continuum Deluxe Reissue", "Album", "John Mayer"), ] assert _best_release_group("John Mayer", "Continuum", groups)["title"] == "Continuum" def test_below_threshold_returns_none(): groups = [_rg("Something Totally Different", "Album", "Michael Jackson")] assert _best_release_group("Michael Jackson", "Off the Wall", groups) is None def test_empty_groups_returns_none(): assert _best_release_group("Michael Jackson", "Off the Wall", []) is None def test_prefers_credited_artist_over_same_titled_other_artist_album(): # Real-world "Fun Machine" bug: Lake Street Dive's EP is what we followed, but two # unrelated bands also have a release group literally titled "Fun Machine" as an Album. # All three tie on title (1.0), so the old is_album tiebreak wrongly grabbed a foreign # Album. The followed artist must win over title-tied albums by other artists. groups = [ _rg("Fun Machine", "EP", "Lake Street Dive"), _rg("Fun Machine", "Album", "Bastards of Melody"), _rg("Fun Machine", "Album", "Teledildonics 5000"), ] best = _best_release_group("Lake Street Dive", "Fun Machine", groups) assert best["artist-credit"][0]["artist"]["name"] == "Lake Street Dive" # --- _with_retry: transient MusicBrainz failures (SSL EOF, 503 rate-limit) --- def test_retry_returns_value_without_backoff_when_first_call_succeeds(): sleeps: list[float] = [] calls = {"n": 0} def fn(): calls["n"] += 1 return "ok" assert _with_retry(fn, sleep=sleeps.append) == "ok" assert calls["n"] == 1 assert sleeps == [] # no backoff when nothing failed def test_retry_recovers_from_transient_network_error(): # The real symptom: arrives # as musicbrainzngs.NetworkError. A couple of retries clears it. sleeps: list[float] = [] calls = {"n": 0} def fn(): calls["n"] += 1 if calls["n"] < 3: raise musicbrainzngs.NetworkError("SSL: UNEXPECTED_EOF_WHILE_READING") return "recovered" assert _with_retry(fn, sleep=sleeps.append) == "recovered" assert calls["n"] == 3 assert sleeps == [1.0, 2.0] # exponential backoff before each retry def test_retry_recovers_from_transient_response_error(): # MusicBrainz's own 503 rate-limit surfaces as ResponseError. calls = {"n": 0} def fn(): calls["n"] += 1 if calls["n"] < 2: raise musicbrainzngs.ResponseError("503 Service Unavailable") return "ok" assert _with_retry(fn, sleep=lambda _: None) == "ok" assert calls["n"] == 2 def test_retry_reraises_after_exhausting_attempts(): sleeps: list[float] = [] def fn(): raise musicbrainzngs.NetworkError("still down") with pytest.raises(musicbrainzngs.NetworkError): _with_retry(fn, attempts=4, sleep=sleeps.append) assert sleeps == [1.0, 2.0, 4.0] # 3 backoffs across 4 attempts def test_retry_does_not_swallow_non_transient_errors(): calls = {"n": 0} def fn(): calls["n"] += 1 raise ValueError("bug, not a network blip") with pytest.raises(ValueError): _with_retry(fn, sleep=lambda _: None) assert calls["n"] == 1 # non-transient errors propagate immediately, no retry