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),
            },
        }


@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"},
    }