mirror of
https://github.com/Dodelidoo-Labs/sonobarr.git
synced 2026-04-18 05:08:21 -04:00
270 lines
9.3 KiB
Python
270 lines
9.3 KiB
Python
"""Tests for Last.fm and ListenBrainz integration services."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from sonobarr_app.services.integrations.lastfm_user import LastFmUserArtist, LastFmUserService
|
|
from sonobarr_app.services.integrations.listenbrainz_user import (
|
|
ListenBrainzIntegrationError,
|
|
ListenBrainzUserService,
|
|
)
|
|
|
|
|
|
class _LBResponse:
|
|
"""Minimal HTTP response object for ListenBrainz tests."""
|
|
|
|
def __init__(self, status_code: int, payload):
|
|
self.status_code = status_code
|
|
self._payload = payload
|
|
|
|
def json(self):
|
|
if isinstance(self._payload, Exception):
|
|
raise self._payload
|
|
return self._payload
|
|
|
|
|
|
class _LBSession:
|
|
"""Simple request session double keyed by URL fragments."""
|
|
|
|
def __init__(self, responses):
|
|
self._responses = responses
|
|
|
|
def get(self, url, timeout):
|
|
for key, value in self._responses.items():
|
|
if key in url:
|
|
return value
|
|
raise AssertionError(f"Unexpected URL: {url}")
|
|
|
|
|
|
def test_lastfm_top_artists_and_recommendations(monkeypatch):
|
|
"""Service should parse top artists and dedupe recommendations from similar artists."""
|
|
|
|
top_entries = [
|
|
SimpleNamespace(item=SimpleNamespace(name="A"), weight="10"),
|
|
SimpleNamespace(item=SimpleNamespace(name="B"), weight="5"),
|
|
]
|
|
|
|
rel_a = [
|
|
SimpleNamespace(item=SimpleNamespace(name="C"), match="0.9"),
|
|
SimpleNamespace(item=SimpleNamespace(name="A"), match="0.8"),
|
|
]
|
|
rel_b = [
|
|
SimpleNamespace(item=SimpleNamespace(name="D"), match="0.7"),
|
|
SimpleNamespace(item=SimpleNamespace(name="C"), match="0.6"),
|
|
]
|
|
|
|
class _ArtistResolver:
|
|
def __init__(self, mapping):
|
|
self.mapping = mapping
|
|
|
|
def get_similar(self):
|
|
return self.mapping
|
|
|
|
network = SimpleNamespace(
|
|
get_user=lambda username: SimpleNamespace(get_top_artists=lambda limit: top_entries),
|
|
get_artist=lambda name: _ArtistResolver(rel_a if name == "A" else rel_b),
|
|
)
|
|
|
|
service = LastFmUserService("key", "secret")
|
|
monkeypatch.setattr(service, "_client", lambda: network)
|
|
|
|
top = service.get_top_artists("user", limit=10)
|
|
recs = service.get_recommended_artists("user", limit=10)
|
|
|
|
assert [item.name for item in top] == ["A", "B"]
|
|
assert [item.playcount for item in top] == [10, 5]
|
|
assert [item.name for item in recs] == ["C", "D"]
|
|
|
|
|
|
def test_lastfm_handles_invalid_payloads():
|
|
"""Service helper methods should remain defensive for malformed relation objects."""
|
|
|
|
service = LastFmUserService("key", "secret")
|
|
bad_name, bad_score = service._parse_similarity_candidate(object())
|
|
assert bad_name == ""
|
|
assert bad_score is None
|
|
|
|
assert service.get_top_artists("", limit=5) == []
|
|
assert service.get_recommended_artists("", limit=5) == []
|
|
|
|
|
|
def test_lastfm_client_and_recommendation_edge_paths(monkeypatch):
|
|
"""Last.fm helper methods should handle constructor wiring and malformed top entries."""
|
|
|
|
created_kwargs = {}
|
|
|
|
class _FakeNetwork:
|
|
def __init__(self, **kwargs):
|
|
created_kwargs.update(kwargs)
|
|
|
|
def get_artist(self, _name):
|
|
raise RuntimeError("boom")
|
|
|
|
monkeypatch.setattr("sonobarr_app.services.integrations.lastfm_user.pylast.LastFMNetwork", _FakeNetwork)
|
|
|
|
service = LastFmUserService("k", "s")
|
|
network = service._client()
|
|
assert created_kwargs == {"api_key": "k", "api_secret": "s"}
|
|
assert service._safe_get_similar(network, "Any") == []
|
|
|
|
top_entries = [SimpleNamespace(item=SimpleNamespace(name=""))]
|
|
assert service._collect_recommendations(network, top_entries, set(), limit=1) == []
|
|
|
|
|
|
def test_lastfm_recommendations_fallback_to_empty_on_transport_failure(monkeypatch):
|
|
"""Recommendation fetch should return an empty list when provider calls fail."""
|
|
|
|
service = LastFmUserService("key", "secret")
|
|
monkeypatch.setattr(service, "_client", lambda: (_ for _ in ()).throw(RuntimeError("network down")))
|
|
assert service.get_recommended_artists("user", limit=20) == []
|
|
|
|
|
|
def test_lastfm_collect_recommendations_returns_early_at_limit():
|
|
"""Recommendation aggregation should stop once the configured limit has been reached."""
|
|
|
|
rel = [
|
|
SimpleNamespace(item=SimpleNamespace(name="Candidate A"), match="0.9"),
|
|
SimpleNamespace(item=SimpleNamespace(name="Candidate B"), match="0.8"),
|
|
]
|
|
|
|
class _Network:
|
|
def get_artist(self, _name):
|
|
return SimpleNamespace(get_similar=lambda: rel)
|
|
|
|
top_entries = [SimpleNamespace(item=SimpleNamespace(name="Seed Artist"))]
|
|
service = LastFmUserService("k", "s")
|
|
recs = service._collect_recommendations(_Network(), top_entries, {"Seed Artist"}, limit=1)
|
|
assert len(recs) == 1
|
|
|
|
|
|
def test_listenbrainz_weekly_exploration_flow():
|
|
"""Service should extract weekly exploration playlist artists and dedupe names."""
|
|
|
|
list_payload = {
|
|
"playlists": [
|
|
{
|
|
"playlist": {
|
|
"identifier": ["https://listenbrainz.org/playlist/abc123/"],
|
|
"extension": {
|
|
"https://musicbrainz.org/doc/jspf#playlist": {
|
|
"additional_metadata": {
|
|
"algorithm_metadata": {
|
|
"source_patch": "weekly-exploration",
|
|
}
|
|
}
|
|
}
|
|
},
|
|
}
|
|
}
|
|
]
|
|
}
|
|
|
|
playlist_payload = {
|
|
"playlist": {
|
|
"track": [
|
|
{
|
|
"extension": {
|
|
"https://musicbrainz.org/doc/jspf#track": {
|
|
"additional_metadata": {
|
|
"artists": [
|
|
{"artist_credit_name": "Artist One"},
|
|
{"name": "Artist Two"},
|
|
]
|
|
}
|
|
}
|
|
}
|
|
},
|
|
{"creator": "Artist One"},
|
|
]
|
|
}
|
|
}
|
|
|
|
session = _LBSession(
|
|
{
|
|
"createdfor": _LBResponse(200, list_payload),
|
|
"/playlist/abc123": _LBResponse(200, playlist_payload),
|
|
}
|
|
)
|
|
service = ListenBrainzUserService(session=session)
|
|
|
|
result = service.get_weekly_exploration_artists("listener")
|
|
|
|
assert result.artists == ["Artist One", "Artist Two"]
|
|
|
|
|
|
def test_listenbrainz_validation_and_error_paths():
|
|
"""Service should return empty results for blank usernames and raise for invalid payloads."""
|
|
|
|
service = ListenBrainzUserService(session=_LBSession({}))
|
|
assert service.get_weekly_exploration_artists(" ").artists == []
|
|
|
|
broken_session = _LBSession({"createdfor": _LBResponse(500, {})})
|
|
broken = ListenBrainzUserService(session=broken_session)
|
|
with pytest.raises(ListenBrainzIntegrationError):
|
|
broken._find_weekly_exploration_playlist("user")
|
|
|
|
invalid_json_session = _LBSession(
|
|
{"createdfor": _LBResponse(200, json.JSONDecodeError("bad", "{}", 0))}
|
|
)
|
|
invalid = ListenBrainzUserService(session=invalid_json_session)
|
|
with pytest.raises(ListenBrainzIntegrationError):
|
|
invalid._find_weekly_exploration_playlist("user")
|
|
|
|
assert ListenBrainzUserService._normalise_identifier(["https://a/b/c/"]) == "c"
|
|
assert ListenBrainzUserService._normalise_identifier(None) == ""
|
|
|
|
|
|
def test_listenbrainz_playlist_discovery_and_parse_edge_paths():
|
|
"""ListenBrainz should skip non-weekly playlists and surface playlist payload JSON errors."""
|
|
|
|
no_weekly_payload = {
|
|
"playlists": [
|
|
{
|
|
"playlist": {
|
|
"identifier": ["https://listenbrainz.org/playlist/not-weekly/"],
|
|
"extension": {
|
|
"https://musicbrainz.org/doc/jspf#playlist": {
|
|
"additional_metadata": {
|
|
"algorithm_metadata": {
|
|
"source_patch": "other-feed",
|
|
}
|
|
}
|
|
}
|
|
},
|
|
}
|
|
}
|
|
]
|
|
}
|
|
no_weekly_service = ListenBrainzUserService(
|
|
session=_LBSession({"createdfor": _LBResponse(200, no_weekly_payload)})
|
|
)
|
|
assert no_weekly_service.get_weekly_exploration_artists("listener").artists == []
|
|
|
|
missing_identifier_payload = {
|
|
"playlists": [
|
|
{
|
|
"playlist": {
|
|
"extension": {
|
|
"https://musicbrainz.org/doc/jspf#playlist": {
|
|
"additional_metadata": {"algorithm_metadata": {"source_patch": "weekly-exploration"}}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
]
|
|
}
|
|
missing_identifier = ListenBrainzUserService(
|
|
session=_LBSession({"createdfor": _LBResponse(200, missing_identifier_payload)})
|
|
)
|
|
assert missing_identifier._find_weekly_exploration_playlist("listener") is None
|
|
|
|
invalid_playlist_json = ListenBrainzUserService(
|
|
session=_LBSession({"playlist/abc": _LBResponse(200, json.JSONDecodeError("bad", "{}", 0))})
|
|
)
|
|
with pytest.raises(ListenBrainzIntegrationError):
|
|
invalid_playlist_json._fetch_playlist_artists("abc")
|