Emma Thorpe
40dfce8a4c
Build and publish container / build (pull_request) Successful in 2m53s
Tag lookups asked by MusicBrainz id whenever Lidarr had one, which is always. The reasoning was that an id cannot be ambiguous the way a name can. In practice Last.fm's mbid index is stale and partial, and it answered "the artist you supplied could not be found" for Devo, Escape the Fate, Blasterjaxx, Frank Carter & the Rattlesnakes and a long tail of others -- artists whose Last.fm pages plainly exist and carry precisely the tags the moods are built from. Devo is tagged new wave, post-punk and 80s; Escape the Fate is tagged post-hardcore, screamo and emocore. Both were skipped. Ask by name first, which is the index Last.fm's own site runs on, and keep the id only as a fallback for a name Lidarr spells differently. Failures were also being dropped without recording the attempt, so every one of those artists was re-queried on every subsequent pass, indefinitely. They are now distinguished: an artist that neither key resolves is recorded as fetched with no tags and not asked about again, while a genuine failure -- a rate limit, a bad key -- is deliberately left unrecorded so the next pass retries it. Telling the two apart needed the service's own error number, so LastfmError now carries it.
307 lines
12 KiB
Python
307 lines
12 KiB
Python
import http.server
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import threading
|
|
import urllib.error
|
|
import urllib.parse
|
|
|
|
import pytest
|
|
|
|
# Ensure the project root is on sys.path when running tests.
|
|
ROOT = os.path.dirname(os.path.dirname(__file__))
|
|
if ROOT not in sys.path:
|
|
sys.path.insert(0, ROOT)
|
|
|
|
|
|
def make_tracks(count, start=1_600_000_000, step=300, artists=5):
|
|
"""Return a synthetic recent-tracks history, oldest first.
|
|
|
|
Mirrors the real payload's quirks: artist and album are sub-objects keyed
|
|
`#text`, MBIDs are present-but-empty rather than absent when unknown, and
|
|
only some entries carry one.
|
|
"""
|
|
return [
|
|
{
|
|
"artist": {"#text": f"Artist {index % artists}", "mbid": f"artist-{index % artists}"},
|
|
"album": {"#text": f"Album {index % 7}", "mbid": ""},
|
|
"name": f"Track {index}",
|
|
"mbid": f"recording-{index}" if index % 3 == 0 else "",
|
|
"date": {"uts": str(start + index * step), "#text": "whenever"},
|
|
}
|
|
for index in range(count)
|
|
]
|
|
|
|
|
|
def make_loved(names):
|
|
"""Return loved-track entries, which key the artist as `name` not `#text`."""
|
|
return [
|
|
{
|
|
"name": name,
|
|
"mbid": "",
|
|
"artist": {"name": "Artist 0", "mbid": "artist-0"},
|
|
"date": {"uts": "1600000000"},
|
|
}
|
|
for name in names
|
|
]
|
|
|
|
|
|
class FakeLastfm:
|
|
"""A transport serving a canned history, so the tests need no network.
|
|
|
|
Honours `from`, `to`, `limit` and `page` the way the real service does, and
|
|
reproduces the two shapes that catch clients out: a lone result comes back
|
|
as a bare object rather than a one-item list, and a currently-playing track
|
|
is prepended to the first page with no `date`.
|
|
"""
|
|
|
|
def __init__(self, tracks=(), loved=(), nowplaying=None, outcomes=(), tags=None):
|
|
# Keyed by mbid or by artist name, whichever the caller asked with.
|
|
self.tags = tags or {}
|
|
self.tracks = sorted(tracks, key=lambda track: int(track["date"]["uts"]), reverse=True)
|
|
self.loved = list(loved)
|
|
self.nowplaying = nowplaying
|
|
# Served in order before any real response; an Exception is raised.
|
|
self.outcomes = list(outcomes)
|
|
self.calls = []
|
|
|
|
def __call__(self, url, timeout=None):
|
|
query = {
|
|
key: value[0]
|
|
for key, value in urllib.parse.parse_qs(urllib.parse.urlparse(url).query).items()
|
|
}
|
|
self.calls.append(query)
|
|
|
|
if self.outcomes:
|
|
outcome = self.outcomes.pop(0)
|
|
if isinstance(outcome, Exception):
|
|
raise outcome
|
|
return json.dumps(outcome)
|
|
|
|
method = query["method"].lower()
|
|
if method == "user.getrecenttracks":
|
|
return json.dumps(self._recent(query))
|
|
if method == "user.getlovedtracks":
|
|
return json.dumps(self._loved(query))
|
|
if method == "artist.gettoptags":
|
|
key = query.get("mbid") or query.get("artist", "")
|
|
if key not in self.tags:
|
|
# What the real service says for a key it cannot resolve, which
|
|
# for mbids is a great many artists whose pages plainly exist.
|
|
return json.dumps(
|
|
{"error": 6, "message": "The artist you supplied could not be found"}
|
|
)
|
|
return json.dumps({"toptags": {"tag": self.tags[key]}})
|
|
raise AssertionError(f"unexpected method {method}")
|
|
|
|
def _recent(self, query):
|
|
selected = self.tracks
|
|
if "to" in query:
|
|
selected = [t for t in selected if int(t["date"]["uts"]) <= int(query["to"])]
|
|
if "from" in query:
|
|
selected = [t for t in selected if int(t["date"]["uts"]) >= int(query["from"])]
|
|
|
|
window = self._page(selected, query)
|
|
if int(query.get("page", 1)) == 1 and self.nowplaying is not None:
|
|
window = [self.nowplaying, *window]
|
|
return {"recenttracks": self._wrap(window, selected, query)}
|
|
|
|
def _loved(self, query):
|
|
window = self._page(self.loved, query)
|
|
return {"lovedtracks": self._wrap(window, self.loved, query)}
|
|
|
|
@staticmethod
|
|
def _page(items, query):
|
|
limit = int(query.get("limit", 50))
|
|
page = int(query.get("page", 1))
|
|
start = (page - 1) * limit
|
|
return items[start : start + limit]
|
|
|
|
@staticmethod
|
|
def _wrap(window, selected, query):
|
|
limit = int(query.get("limit", 50))
|
|
total = len(selected)
|
|
return {
|
|
# A single result is not wrapped in a list by the real service.
|
|
"track": window[0] if len(window) == 1 else window,
|
|
"@attr": {
|
|
"user": query.get("user", ""),
|
|
"page": query.get("page", "1"),
|
|
"perPage": str(limit),
|
|
"totalPages": str(max(1, -(-total // limit))),
|
|
"total": str(total),
|
|
},
|
|
}
|
|
|
|
|
|
class FakeLidarr:
|
|
"""A transport serving a canned library over the Lidarr v1 API surface.
|
|
|
|
Built from a nested description -- artist, album, tracks -- and split back
|
|
out across the four endpoints the indexer actually calls, so the tests
|
|
exercise the same stitching the real thing does.
|
|
"""
|
|
|
|
def __init__(self, artists=(), fail=()):
|
|
# (path, artistId) pairs the fake refuses to serve, standing in for the
|
|
# 500s Lidarr returns on data it cannot hydrate.
|
|
self.fail = set(fail)
|
|
self.artists, self.albums, self.tracks, self.files = [], [], [], []
|
|
for artist_index, entry in enumerate(artists, start=1):
|
|
artist_id = artist_index
|
|
self.artists.append(
|
|
{
|
|
"id": artist_id,
|
|
"artistName": entry["name"],
|
|
"foreignArtistId": entry.get("mbid", f"artist-mbid-{artist_id}"),
|
|
"path": f"/music/{entry['name']}",
|
|
"monitored": entry.get("monitored", True),
|
|
}
|
|
)
|
|
for album_index, album in enumerate(entry.get("albums", []), start=1):
|
|
album_id = artist_id * 100 + album_index
|
|
self.albums.append(
|
|
{
|
|
"id": album_id,
|
|
"artistId": artist_id,
|
|
"title": album["title"],
|
|
"foreignAlbumId": f"album-mbid-{album_id}",
|
|
"monitored": album.get("monitored", True),
|
|
"releaseDate": album.get("release_date", "2019-01-01T00:00:00Z"),
|
|
}
|
|
)
|
|
for track_index, track in enumerate(album.get("tracks", []), start=1):
|
|
track_id = album_id * 100 + track_index
|
|
has_file = track.get("has_file", True)
|
|
self.tracks.append(
|
|
{
|
|
"id": track_id,
|
|
"artistId": artist_id,
|
|
"albumId": album_id,
|
|
"title": track["title"],
|
|
"foreignRecordingId": track.get("recording_mbid", ""),
|
|
"trackFileId": track_id if has_file else 0,
|
|
"hasFile": has_file,
|
|
"duration": 210000,
|
|
}
|
|
)
|
|
if has_file:
|
|
self.files.append(
|
|
{
|
|
"id": track_id,
|
|
"artistId": artist_id,
|
|
"albumId": album_id,
|
|
"path": f"/music/{entry['name']}/{album['title']}/"
|
|
f"{track['title']}.flac",
|
|
"dateAdded": track.get("added", "2020-05-01T12:00:00Z"),
|
|
}
|
|
)
|
|
self.calls = []
|
|
|
|
def __call__(self, url, timeout=None, headers=None):
|
|
parsed = urllib.parse.urlparse(url)
|
|
assert (headers or {}).get("X-Api-Key"), "Lidarr requires the API key header"
|
|
path = parsed.path.rsplit("/", 1)[-1]
|
|
query = {
|
|
key: value[0] for key, value in urllib.parse.parse_qs(parsed.query).items()
|
|
}
|
|
self.calls.append((path, query))
|
|
|
|
artist_id = int(query.get("artistId", 0))
|
|
if (path, artist_id) in self.fail:
|
|
raise urllib.error.HTTPError(
|
|
url, 500, "Internal Server Error", {}, io.BytesIO(b'{"message": "boom"}')
|
|
)
|
|
|
|
album_ids = query.get("albumIds")
|
|
if path == "album" and album_ids:
|
|
album_id = int(album_ids)
|
|
if ("albumid", album_id) in self.fail:
|
|
raise urllib.error.HTTPError(
|
|
url,
|
|
500,
|
|
"Internal Server Error",
|
|
{},
|
|
io.BytesIO(b'{"message": "Sequence contains more than one element"}'),
|
|
)
|
|
return json.dumps([row for row in self.albums if row["id"] == album_id])
|
|
|
|
if path == "artist":
|
|
return json.dumps(self.artists)
|
|
source = {"album": self.albums, "track": self.tracks, "trackfile": self.files}[path]
|
|
if path == "album" and not artist_id:
|
|
# Lidarr's unfiltered album endpoint returns the lot.
|
|
return json.dumps(source)
|
|
if not artist_id:
|
|
raise urllib.error.HTTPError(
|
|
url,
|
|
400,
|
|
"Bad Request",
|
|
{},
|
|
io.BytesIO(b'{"message": "artistId must be provided"}'),
|
|
)
|
|
return json.dumps([row for row in source if row["artistId"] == artist_id])
|
|
|
|
|
|
@pytest.fixture
|
|
def now_playing():
|
|
"""The entry Last.fm prepends for a track in progress: no `date` at all."""
|
|
return {
|
|
"artist": {"#text": "Artist 0", "mbid": "artist-0"},
|
|
"album": {"#text": "Album 0", "mbid": ""},
|
|
"name": "Currently Playing",
|
|
"mbid": "",
|
|
"@attr": {"nowplaying": "true"},
|
|
}
|
|
|
|
|
|
class _CountingServer(http.server.ThreadingHTTPServer):
|
|
"""Counts accepted connections, which is what connection reuse is about."""
|
|
|
|
daemon_threads = True
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
self.connections = 0
|
|
super().__init__(*args, **kwargs)
|
|
|
|
def process_request(self, request, client_address):
|
|
self.connections += 1
|
|
super().process_request(request, client_address)
|
|
|
|
|
|
class _Handler(http.server.BaseHTTPRequestHandler):
|
|
# Without HTTP/1.1 the server closes after every response and no client
|
|
# could reuse anything, which would make the test prove nothing.
|
|
protocol_version = "HTTP/1.1"
|
|
|
|
def do_GET(self):
|
|
if self.path.startswith("/boom"):
|
|
body, status = b'{"message": "boom"}', 500
|
|
else:
|
|
body = json.dumps(
|
|
{"path": self.path, "key": self.headers.get("X-Api-Key")}
|
|
).encode()
|
|
status = 200
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass
|
|
|
|
|
|
@pytest.fixture
|
|
def http_server():
|
|
"""A real local HTTP server, for the one component that talks sockets."""
|
|
server = _CountingServer(("127.0.0.1", 0), _Handler)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
try:
|
|
yield server, f"http://127.0.0.1:{server.server_port}"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|