import io import json import os import sys 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=()): 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)) 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"}') ) 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"}, }