muse/tests/test_acquisition.py

"""The acquisition ladder: query variants, candidate fallback, stall handling.

This is the path that had no coverage at all, and it is where every production
failure lived. A fake slskd stands in for the network so the branches that only
happen on a bad day -- a peer that rejects, a transfer slskd forgets, a search
the server filters -- can be exercised deterministically.

Run standalone: python tests/test_acquisition.py
"""
import asyncio, json, os, sys

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/acquisition.db")
os.environ.setdefault("MUSE_SEARCH_POLLS", "2")

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

import app  # noqa: E402

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


# Searches poll on a 2s cadence; nothing here is actually waiting on a network.
async def _instant(_):
    return None


app.asyncio.sleep = _instant
app.init_db()          # normally done in lifespan; there is no server here


# ── a slskd that behaves however the test needs ──────────────────────────

class Resp:
    def __init__(self, data=None, status=200):
        self._data, self.status = data, status

    def raise_for_status(self):
        if self.status >= 400:
            raise RuntimeError(f"Server error '{self.status} Internal Server Error'")

    def json(self):
        return self._data


class FakeSlskd:
    """Records what was asked of it and answers from a script."""

    def __init__(self, per_query=None, enqueue_status=None, transfers=None):
        self.per_query = per_query or {}      # query text -> list of responses
        self.enqueue_status = enqueue_status or {}   # username -> http status
        self.transfers = transfers or []
        self.searched = []
        self.enqueued = []
        self.deleted = []
        self._last = None

    def __call__(self, timeout=30.0):
        return self

    async def __aenter__(self):
        return self

    async def __aexit__(self, *a):
        return False

    async def post(self, url, json=None):
        if url == "/api/v0/searches":
            self._last = json["searchText"]
            self.searched.append(self._last)
            return Resp({})
        user = url.rsplit("/", 1)[-1]
        st = self.enqueue_status.get(user, 200)
        self.enqueued.append(user)
        return Resp({}, st)

    async def get(self, url, params=None):
        if url.startswith("/api/v0/searches/"):
            responses = self.per_query.get(self._last, [])
            return Resp({"state": "Completed, ResponseLimitReached",
                         "responses": responses,
                         "fileCount": sum(len(r["files"]) for r in responses)})
        return Resp(self.transfers)

    async def delete(self, url, params=None):
        self.deleted.append(url)
        return Resp({})


def peer(username, filename, *, slot=True, queue=0, size=5_000_000, bitrate=320):
    return {"username": username, "hasFreeUploadSlot": slot,
            "queueLength": queue, "uploadSpeed": 1_000_000,
            "files": [{"filename": filename, "size": size, "bitRate": bitrate}]}


def make_job(job_id, artist="Michael Jackson", title="Billie Jean"):
    with app.db() as c:
        c.execute("INSERT OR REPLACE INTO jobs (id, state, artist, title, album)"
                  " VALUES (?,?,?,?,?)", (job_id, "searching", artist, title, None))


def job_row(job_id):
    with app.db() as c:
        return c.execute("SELECT * FROM jobs WHERE id=?", (job_id,)).fetchone()


def main() -> int:
    print("\nnormalisation")
    check("accents fold to ASCII", app.norm("Beyoncé") == "beyonce",
          repr(app.norm("Beyoncé")))
    check("umlauts fold", app.norm("Björk") == "bjork", repr(app.norm("Björk")))
    check("slashed o folds", app.norm("margø") == "margo", repr(app.norm("margø")))
    check("apostrophes vanish, not split",
          app.norm("Don't Stop") == "dont stop", repr(app.norm("Don't Stop")))
    check("ampersand still separates", app.norm("R&B") == "r b", repr(app.norm("R&B")))

    print("\nquery ladder")
    qs = app.search_queries("Michael Jackson", "Billie Jean")
    check("falls back to the bare title", "Billie Jean" in qs, qs)
    check("full query is still tried first", qs[0] == "Michael Jackson Billie Jean", qs)
    qs = app.search_queries("Frank Ocean", "Novacane (Explicit Version)")
    check("strips a parenthetical suffix", "Frank Ocean Novacane" in qs, qs)
    qs = app.search_queries("Post Sex Nachos", "Pop Tarts (feat. MARIS)")
    check("strips a featuring credit", "Post Sex Nachos Pop Tarts" in qs, qs)
    qs = app.search_queries("The Weeknd", "House Of Balloons / Glass Table Girls")
    check("splits a slashed title", "House Of Balloons" in qs, qs)
    check("ladder is capped", len(app.search_queries("A", "B")) <= app.MAX_QUERIES)

    print("\nscoring")
    avail_mp3 = app.score({"filename": r"x\Radiohead - Creep.mp3", "size": 1,
                           "bitRate": 320},
                          {"username": "u", "hasFreeUploadSlot": True,
                           "queueLength": 0, "uploadSpeed": 0},
                          "Radiohead", "Creep")
    queued_flac = app.score({"filename": r"x\Radiohead - Creep.flac", "size": 1,
                             "bitRate": 1000},
                            {"username": "v", "hasFreeUploadSlot": False,
                             "queueLength": 400, "uploadSpeed": 0},
                            "Radiohead", "Creep")
    check("an available MP3 beats a FLAC behind 400 people",
          avail_mp3 > queued_flac, f"{avail_mp3:.0f} vs {queued_flac:.0f}")
    check("artist found in the folder, not the filename",
          app.match_ratio(r"Radiohead\Pablo Honey\02 Creep.flac",
                          "Radiohead", "Creep") == 1.0)
    check("a matching artist cannot rescue the wrong song",
          app.match_ratio(r"Radiohead\Pablo Honey\04 Karma Police.flac",
                          "Radiohead", "Creep") == 0.0)
    check("reordered filename still matches",
          app.match_ratio(r"x\Creep - Radiohead.mp3", "Radiohead", "Creep") == 1.0)

    print("\ncandidate ladder")
    fake = FakeSlskd(per_query={
        "Billie Jean": [peer("alice", r"m\Michael Jackson\Billie Jean.flac"),
                        peer("bob", r"n\Michael Jackson\Billie Jean.mp3"),
                        peer("carol", r"o\Michael Jackson\Billie Jean.flac")],
    })
    app.slskd_client = fake
    make_job("j1")
    asyncio.run(app.pick_and_enqueue("j1", "Michael Jackson", "Billie Jean"))
    r = job_row("j1")
    check("blocklisted artist recovered by the title rung",
          "Billie Jean" in fake.searched, fake.searched)
    check("job is downloading", r["state"] == "downloading", r["detail"])
    check("runners-up were kept", len(json.loads(r["candidates"] or "[]")) == 2,
          r["candidates"])
    check("consecutive candidates are different peers",
          len({c["username"] for c in json.loads(r["candidates"])}) == 2)

    print("\nfallback on failure")
    fake.enqueue_status = {}
    asyncio.run(app.abandon(job_row("j1"), "Transfer rejected: File not shared."))
    r = job_row("j1")
    check("a rejection advances instead of failing",
          r["state"] == "downloading", f"{r['state']} / {r['detail']}")
    check("the failed peer is recorded",
          "File not shared" in (r["tried"] or ""), r["tried"])
    check("attempt counter advanced", r["attempts"] == 2, r["attempts"])

    asyncio.run(app.abandon(job_row("j1"), "timed out"))
    asyncio.run(app.abandon(job_row("j1"), "timed out"))
    r = job_row("j1")
    check("only fails once the ladder is exhausted", r["state"] == "failed", r["detail"])
    check("failure names every attempt, not just the last",
          "tried 3 peers" in (r["detail"] or ""), r["detail"])
    # A row carried over from before the ladder has attempts=0 but a real
    # failure recorded, which must not render as "tried 0 peers: 1x ...".
    check("a migrated row counts its evidence, not its counter",
          app.exhausted_detail(
              [{"username": "u1", "reason": "no bytes in 20m"}], 0)
          == "tried 1 peer: 1x no bytes in 20m",
          app.exhausted_detail([{"username": "u1", "reason": "no bytes in 20m"}], 0))

    print("\nattempt cap")
    fake2 = FakeSlskd(per_query={"Creep": [
        peer(f"u{i}", rf"x\Radiohead\Creep.flac") for i in range(8)]})
    # every peer refuses to accept the transfer
    fake2.enqueue_status = {f"u{i}": 500 for i in range(8)}
    app.slskd_client = fake2
    # The cap defaults to the ladder size, so lower it here or this only ever
    # proves the list ran out -- which is a different branch.
    cap, app.MAX_ATTEMPTS = app.MAX_ATTEMPTS, 3
    make_job("j2", "Radiohead", "Creep")
    asyncio.run(app.pick_and_enqueue("j2", "Radiohead", "Creep"))
    r = job_row("j2")
    check("gives up at MAX_ATTEMPTS, not at the end of the list",
          r["state"] == "failed" and r["attempts"] == 3,
          f"state={r['state']} attempts={r['attempts']}")
    check("and leaves the untried remainder alone",
          len(json.loads(r["candidates"] or "[]")) > 0, r["candidates"])
    app.MAX_ATTEMPTS = cap
    check("unreachable peers are reported as such",
          "could not reach peer" in (r["tried"] or ""), r["detail"])

    print("\nsystemic failures do not burn the ladder")
    # slskd timing out says nothing about any particular file, so walking the
    # whole list to learn it once is waste -- and it is what took a real
    # request down: five enqueues in twenty-five seconds during one slskd stall.
    fake4 = FakeSlskd(per_query={"Chicago": [
        peer(f"p{i}", r"x\Michael Jackson\Xscape\Chicago.flac") for i in range(8)]})

    class Timeouts(FakeSlskd):
        async def post(self, url, json=None):
            if url != "/api/v0/searches":
                raise RuntimeError("The wait timed out after 5000 milliseconds")
            return await FakeSlskd.post(self, url, json=json)

    t = Timeouts(per_query=fake4.per_query)
    app.slskd_client = t
    make_job("j5", "Michael Jackson", "Chicago")
    asyncio.run(app.pick_and_enqueue("j5", "Michael Jackson", "Chicago"))
    r = job_row("j5")
    check("stops after SYSTEMIC_LIMIT instead of walking the whole list",
          r["attempts"] == app.SYSTEMIC_LIMIT, r["attempts"])
    left = json.loads(r["candidates"] or "[]")
    check("untried candidates are preserved for a retry", len(left) > 0, len(left))
    check("the message says a retry will resume",
          "retry resumes" in (r["detail"] or ""), r["detail"])

    print("\nempty results")
    fake3 = FakeSlskd(per_query={})
    app.slskd_client = fake3
    make_job("j3", "Nobody", "Nothing")
    asyncio.run(app.pick_and_enqueue("j3", "Nobody", "Nothing"))
    r = job_row("j3")
    check("says the search was empty, not that matching failed",
          "Soulseek returned nothing" in (r["detail"] or ""), r["detail"])
    check("names the queries it tried", "'Nothing'" in (r["detail"] or ""), r["detail"])

    print("\ntransient slskd")
    class Flaky(FakeSlskd):
        def __init__(self):
            super().__init__(per_query={"Creep": [peer("z", r"x\Creep.flac")]})
            self.n = 0

        async def post(self, url, json=None):
            if url == "/api/v0/searches":
                self.n += 1
                if self.n < 3:
                    raise RuntimeError("")   # httpx timeouts stringify to nothing
            return await FakeSlskd.post(self, url, json=json)

    flaky = Flaky()
    app.slskd_client = flaky
    make_job("j4", "Radiohead", "Creep")
    asyncio.run(app.pick_and_enqueue("j4", "Radiohead", "Creep"))
    check("a blip is retried rather than failing the job",
          job_row("j4")["state"] == "downloading", job_row("j4")["detail"])

    print("\nsearch hygiene")
    check("finished searches are cleaned up", len(flaky.deleted) >= 1)

    print("\nstall clock")
    from datetime import datetime, timedelta, timezone
    with app.db() as c:
        c.execute("UPDATE jobs SET started_at=? WHERE id=?",
                  ((datetime.now(timezone.utc) - timedelta(minutes=90)).isoformat(),
                   "j4"))
    check("a 90-minute-old transfer reads as stalled",
          app.stalled_for(job_row("j4")) > app.STALL_MINUTES,
          app.stalled_for(job_row("j4")))
    # A row written before started_at existed still has to age out, or the
    # jobs that have been stuck longest are the only ones exempt from the check.
    with app.db() as c:
        c.execute("UPDATE jobs SET started_at=NULL, created_at=? WHERE id=?",
                  ((datetime.now(timezone.utc) - timedelta(days=2))
                   .isoformat(sep=" ")[:19], "j4"))
    check("a pre-upgrade row falls back to created_at",
          app.stalled_for(job_row("j4")) > app.STALL_MINUTES,
          app.stalled_for(job_row("j4")))

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


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