music-curator/tests/conftest.py

import json
import os
import sys
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=()):
        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))

        if path == "artist":
            return json.dumps(self.artists)
        artist_id = int(query.get("artistId", 0))
        source = {"album": self.albums, "track": self.tracks, "trackfile": self.files}[path]
        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"},
    }