muse/tests/test_playlist.py

"""Playlist import: link resolution, ordering, and the m3u Navidrome reads.

The resolvers are exercised against synthetic payloads rather than the live
sites. Two reasons: a test that fails when Spotify has a bad afternoon is not
testing this code, and the failure that actually matters here is the *shape*
changing -- which is precisely what a fixture can assert and a live fetch
cannot.

Run standalone: python tests/test_playlist.py
"""
import asyncio, json, os, sys
from pathlib import Path

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
os.environ.setdefault("MUSE_AUTH", "off")
os.environ.setdefault("MUSE_DB", "dev/playlist.db")
os.environ.setdefault("MUSE_LIBRARY_DIR", "dev/pl-library")

for f in ("dev/playlist.db", "dev/playlist.db-wal", "dev/playlist.db-shm"):
    if os.path.exists(f):
        os.remove(f)

import app  # noqa: E402

app.init_db()
app.LIBRARY_DIR.mkdir(parents=True, exist_ok=True)

ok = fail = 0


def check(name, cond, detail=""):
    global ok, fail
    if cond:
        ok += 1
        print(f"  PASS  {name}")
    else:
        fail += 1
        print(f"  FAIL  {name}  {detail}")


SPOTIFY_PAGE = (
    '<html><script id="__NEXT_DATA__" type="application/json">'
    + json.dumps({"props": {"pageProps": {"state": {"data": {"entity": {
        "name": "Late Night",
        "trackList": [
            {"title": "The Color Violet", "subtitle": "Tory Lanez"},
            {"title": "No Pole", "subtitle": "Don Toliver"},
        ]}}}}}})
    + "</script></html>")

APPLE_PAGE = (
    '<html><script id="serialized-server-data" type="application/json">'
    + json.dumps([{"data": {"sections": [
        {"headerTitle": "Heavy Rotation"},
        {"items": [
            {"title": "YUKON", "artistName": "Justin Bieber"},
            {"title": "LET THE WORLD BURN", "artistName": "Chris Grey"},
            {"title": "YUKON", "artistName": "Justin Bieber"},   # dupe
        ]}]}}])
    + "</script></html>")


def fake_fetch(body):
    async def _f(url):
        return body
    return _f


def make_playlist(pid, name, rows):
    """rows: (seq, artist, title, state, imported_path)"""
    with app.db() as c:
        c.execute("INSERT OR REPLACE INTO playlists (id,name,source,"
                  "source_url,tracks_total) VALUES (?,?,?,?,?)",
                  (pid, name, "spotify", "https://x", len(rows)))
        for seq, artist, title, state, path in rows:
            c.execute(
                "INSERT INTO jobs (id,state,detail,artist,title,"
                "imported_path,playlist_id,playlist_seq) "
                "VALUES (?,?,?,?,?,?,?,?)",
                (f"{pid}-{seq}", state, "", artist, title, path, pid, seq))


def main() -> int:
    print("\nlink routing")
    for url, want in [
        ("https://open.spotify.com/playlist/abc123", "spotify"),
        ("https://open.spotify.com/intl-pt/playlist/abc123", "spotify"),
        ("https://music.apple.com/us/playlist/hits/pl.abc-123", "apple"),
        ("https://www.deezer.com/en/playlist/1313621735", "deezer"),
    ]:
        matched = ("spotify" if app.SPOTIFY_URL.search(url)
                   else "apple" if app.APPLE_URL.search(url)
                   else "deezer" if app.DEEZER_URL.search(url) else "none")
        check(f"{want} link routes to {want}", matched == want, f"got {matched}")

    print("\nresolvers, against a fixed payload shape")
    real = app._fetch
    try:
        app._fetch = fake_fetch(SPOTIFY_PAGE)
        name, tracks = asyncio.run(app._spotify_playlist("playlist", "x"))
        check("spotify name", name == "Late Night", name)
        check("spotify pulls artist from subtitle",
              tracks[0] == ("Tory Lanez", "The Color Violet"), tracks[:1])

        app._fetch = fake_fetch(APPLE_PAGE)
        name, tracks = asyncio.run(app._apple_playlist("playlist", "pl.x", "u"))
        check("apple name", name == "Heavy Rotation", name)
        check("apple pulls artist from artistName",
              ("Justin Bieber", "YUKON") in tracks, tracks)
        check("apple de-duplicates repeated entries", len(tracks) == 2, tracks)

        # The failure that actually matters: they changed the page.
        app._fetch = fake_fetch("<html>nothing useful here</html>")
        for fn, label in ((app._spotify_playlist, "spotify"),
                          (app._apple_playlist, "apple")):
            try:
                asyncio.run(fn("playlist", "x", "u") if label == "apple"
                            else fn("playlist", "x"))
                check(f"{label} shape change is reported", False, "no raise")
            except app.PlaylistUnavailable as e:
                check(f"{label} shape change names the resolver",
                      "needs updating" in str(e), str(e))
    finally:
        app._fetch = real

    print("\nthe m3u Navidrome reads")
    lib = app.LIBRARY_DIR
    (lib / "Tory Lanez").mkdir(parents=True, exist_ok=True)
    (lib / "Don Toliver").mkdir(parents=True, exist_ok=True)
    a = lib / "Tory Lanez" / "The Color Violet.flac"; a.write_text("x")
    b = lib / "Don Toliver" / "No Pole.flac"; b.write_text("x")

    make_playlist("p1", "Late Night", [
        (1, "Don Toliver", "No Pole", "imported", str(b)),
        (0, "Tory Lanez", "The Color Violet", "imported", str(a)),
        (2, "Nobody", "Missing", "failed", None),
    ])
    dest = app.write_playlist("p1")
    text = dest.read_text()
    check("m3u is written", dest and dest.exists(), dest)
    check("declares the playlist name for Navidrome",
          "#PLAYLIST:Late Night" in text, text)
    check("uses Navidrome's mount, not muse's",
          "/music/Tory Lanez/The Color Violet.flac" in text
          and str(lib) not in text, text)
    check("keeps the playlist's own order, not insertion order",
          text.index("The Color Violet") < text.index("No Pole"), text)
    check("omits tracks that never imported", "Missing" not in text, text)

    print("\nedge cases")
    make_playlist("p2", "Nothing Landed Yet", [
        (0, "A", "B", "searching", None)])
    check("no m3u while the playlist is empty",
          app.write_playlist("p2") is None)
    check("and no stray file left behind",
          not (lib / "Nothing Landed Yet.m3u").exists())

    make_playlist("p3", "../../escape", [
        (0, "Tory Lanez", "The Color Violet", "imported", str(a))])
    out = app.write_playlist("p3")
    check("a traversing playlist name cannot escape the library",
          out is None or out.resolve().is_relative_to(lib.resolve()), out)

    # A track approved and then rejected has to leave the file.
    with app.db() as c:
        c.execute("UPDATE jobs SET state='failed' WHERE id=?", ("p1-0",))
    app.write_playlist("p1")
    check("rejecting a track drops it from the m3u",
          "The Color Violet" not in dest.read_text(), dest.read_text())

    print(f"\n{ok} passed, {fail} failed")
    return 1 if fail else 0


if __name__ == "__main__":
    sys.exit(main())