Emma Thorpe
15e5ee5aea
Build and publish container / build (pull_request) Successful in 3m50s
A second set of playlists selecting by genre and mood rather than by play history: eighties synths, high energy rock, screamo, drum and bass, dance, classic rock. The tags come from Last.fm rather than MusicBrainz. MusicBrainz genres arrive free with the Lidarr index, which makes them the obvious choice and the wrong one: they are sparse and formal, and will not tell you a record is screamo or synthwave. Crowd tags will, because people typed them. One request per artist, by MusicBrainz id where Lidarr has one, refreshed every ninety days. An artist Last.fm has never heard of is recorded as fetched with no tags rather than left unmarked, so it is not asked about again on every pass forever. The tag table is keyed on the normalised artist name, not the Lidarr id, so it survives an artist being removed and re-added there. Weights are taken from the response's count where it has one. The documented sample carries only a name and a URL, a live response also carries a 0-100 count, and depending on either alone would be a guess -- so the count is used when present and the documented ordering by popularity stands in when it is not. An artist qualifies for a mood when their weights inside it sum to at least thirty. A single low-weight tag is not a genre, it is somebody's stray opinion. A mood may also restrict release years, which is what separates eighties synth records from everything else a synthpop tag drags in. The built-in set is chosen for this library rather than as a taxonomy, and --vibes replaces it wholesale with a JSON file so a new mood does not need a new release. Names are validated when that file is read: an invalid one would otherwise only surface as a playlist written somewhere unintended.
301 lines
11 KiB
Python
301 lines
11 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", "")
|
|
return json.dumps({"toptags": {"tag": self.tags.get(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()
|