ani-web/tests/test_api.py

"""HTTP-level tests for ani-web (stdlib only), run via pytest.

Uses the throwaway server + temp data dir from conftest.py — never
touches the real server/data. Seeded users: admin/alice/bob/dave/erin
(pw <name>pw12345).
"""
import http.cookiejar as _cj
import json
import urllib.error
import urllib.request

BASE = "http://127.0.0.1:8399"
CJ = lambda: _cj.CookieJar()


class _NoRedirect(urllib.request.HTTPRedirectHandler):
    """urlopen follows 302s by default; we want to *see* them."""
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        return None


def no_redirect(method, path, data=None, base=BASE):
    """Request without following redirects; returns (status, headers)."""
    opener = urllib.request.build_opener(_NoRedirect)
    r = urllib.request.Request(base + path,
                               data=json.dumps(data).encode() if data is not None else None,
                               method=method,
                               headers={"Content-Type": "application/json", "X-Aniweb": "1"})
    try:
        resp = opener.open(r)
        return resp.getcode(), resp.headers
    except urllib.error.HTTPError as e:
        return e.code, e.headers


def no_redirect_get(path, base=BASE):
    return no_redirect("GET", path, base=base)


def req(method, path, data=None, jar=None, base=BASE):
    r = urllib.request.Request(base + path,
                               data=json.dumps(data).encode() if data is not None else None,
                               method=method,
                               headers={"Content-Type": "application/json", "X-Aniweb": "1"})
    if jar is not None:
        opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
        try:
            return opener.open(r).getcode()
        except urllib.error.HTTPError as e:
            return e.code
    try:
        return urllib.request.urlopen(r).status
    except urllib.error.HTTPError as e:
        return e.code


def login(user, pw, base=BASE):
    jar = CJ()
    assert req("POST", "/login", {"username": user, "password": pw}, jar, base) == 200, user
    return jar


def test_unauthenticated(base=BASE):
    # page -> 302 to /login, api -> 401
    code, headers = no_redirect_get("/", base)
    assert code == 302 and "/login" in headers["Location"], (code, headers.get("Location"))
    assert req("GET", "/api/search?q=x", base=base) == 401


def test_wrong_password(base=BASE):
    assert req("POST", "/login", {"username": "admin", "password": "nope"}, base=base) == 401


def test_login_and_access(base=BASE):
    jar = login("admin", "adminpw12345", base)
    assert req("GET", "/api/search?q=x", jar=jar, base=base) == 200
    assert req("GET", "/api/me", jar=jar, base=base) == 200


def test_logout(base=BASE):
    jar = login("admin", "adminpw12345", base)
    assert req("POST", "/logout", jar=jar, base=base) == 200
    # session gone: cookie jar emptied by the delete-cookie response
    assert not [c for c in jar if c.name == "aniweb_session"], "cookie not cleared"
    # and a cookie-less request is bounced to /login
    code, headers = no_redirect_get("/", base)
    assert code == 302 and "/login" in headers["Location"]


def test_no_self_signup(base=BASE):
    # /signup route removed — unauthenticated POST bounces to /login
    code, headers = no_redirect("POST", "/signup",
                                {"username": "zed", "password": "whatever123"}, base)
    assert code in (302, 404, 405), code
    # and even authenticated it's gone (no such route)
    jar = login("admin", "adminpw12345", base)
    assert req("POST", "/signup", {"username": "zed", "password": "whatever123"},
               jar=jar, base=base) == 405
    assert "zed" not in _users()


def test_history_requires_ep_id(base=BASE):
    jar = login("admin", "adminpw12345", base)
    assert req("POST", "/api/history", {"slug_id": "x"}, jar=jar, base=base) == 422


def test_history_per_user(base=BASE):
    a = login("admin", "adminpw12345", base)
    b = login("alice", "alicepw12345", base)
    entry = {"ep_id": "111", "slug_id": "test-111", "title": "Test", "season_id": 1,
             "ep": "111", "lang": "eng", "pos": 42.0}
    assert req("POST", "/api/history", entry, jar=a, base=base) == 200
    got_a = _json("GET", "/api/history", jar=a, base=base)
    got_b = _json("GET", "/api/history", jar=b, base=base)
    # ep_id is normalized to int on write (ghost-row fix)
    assert any(e["ep_id"] == 111 for e in got_a)
    assert all(e.get("ep_id") != 111 for e in got_b), "leak: admin entry in alice's history"


def test_history_all_and_delete(base=BASE):
    jar = login("alice", "alicepw12345", base)
    e1 = {"ep_id": 201, "slug_id": "del-a", "title": "Del A", "season_id": 1,
          "ep": "1", "lang": "eng", "pos": 5}
    e2 = {"ep_id": "202", "slug_id": "del-b", "title": "Del B", "season_id": 1,
          "ep": "2", "lang": "eng", "pos": 5}
    assert req("POST", "/api/history", e1, jar=jar, base=base) == 200
    assert req("POST", "/api/history", e2, jar=jar, base=base) == 200
    all_ = _json("GET", "/api/history?all=1", jar=jar, base=base)
    assert {e["slug_id"] for e in all_} >= {"del-a", "del-b"}
    # plain GET still caps at 5
    assert len(_json("GET", "/api/history", jar=jar, base=base)) <= 5
    # loose str match: the int ep_id row is deleted via a string query param
    removed = _json("DELETE", "/api/history?slug_id=del-a&ep_id=201", jar=jar, base=base)
    assert removed["removed"] >= 1, removed
    all_ = _json("GET", "/api/history?all=1", jar=jar, base=base)
    assert all(e["slug_id"] != "del-a" for e in all_)
    assert any(e["slug_id"] == "del-b" for e in all_)
    # idempotent
    assert _json("DELETE", "/api/history?slug_id=del-a&ep_id=201",
                 jar=jar, base=base)["removed"] == 0


def test_history_no_ghost_duplicates(base=BASE):
    """same episode posted with int and string ep_id -> exactly one row"""
    jar = login("alice", "alicepw12345", base)
    for ep in (301, "301"):
        assert req("POST", "/api/history",
                   {"ep_id": ep, "slug_id": "ghost-a", "title": "Ghost", "season_id": 1,
                    "ep": "1", "lang": "eng", "pos": 5}, jar=jar, base=base) == 200
    rows = [e for e in _json("GET", "/api/history?all=1", jar=jar, base=base)
            if e["slug_id"] == "ghost-a"]
    assert len(rows) == 1 and rows[0]["ep_id"] == 301, rows


def test_recent_searches(base=BASE):
    a = login("alice", "alicepw12345", base)
    assert req("POST", "/api/searches", {"q": "naruto"}, jar=a, base=base) == 200
    assert req("POST", "/api/searches", {"q": "frieren"}, jar=a, base=base) == 200
    # case-insensitive dedupe, newest first
    assert req("POST", "/api/searches", {"q": "NARUTO"}, jar=a, base=base) == 200
    assert _json("GET", "/api/searches", jar=a, base=base) == ["NARUTO", "frieren"]
    # per-user isolation (browser tests write admin's searches, but only after
    # the API phase, so admin is clean here)
    b = login("admin", "adminpw12345", base)
    assert _json("GET", "/api/searches", jar=b, base=base) == []
    # blank q rejected
    assert req("POST", "/api/searches", {"q": "   "}, jar=a, base=base) == 422


def test_self_delete(base=BASE):
    jar = login("dave", "davepw12345", base)
    # give dave a searches file too: account delete must remove it
    assert req("POST", "/api/searches", {"q": "dave-search"}, jar=jar, base=base) == 200
    assert req("POST", "/account/delete", jar=jar, base=base) == 200
    assert "dave" not in _users()
    assert _search_count("dave") == 0  # account delete cascades the rows
    # old session cookie is dead immediately (middleware re-checks existence)
    assert req("GET", "/api/me", jar=jar, base=base) == 401
    jar2 = CJ()
    assert req("POST", "/login", {"username": "dave", "password": "davepw12345"}, jar2, base=base) == 401


def test_csrf_header_required(base=BASE):
    # mutation routes without the X-Aniweb header -> 403 (a cross-site form
    # POST can't set it; the SPA always sends it). /login is exempt.
    jar = login("admin", "adminpw12345", base)
    r = urllib.request.Request(base + "/logout", method="POST")
    try:
        assert urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(jar)).open(r).status == 403
    except urllib.error.HTTPError as e:
        assert e.code == 403, e.code
    # with the header (what the SPA sends) it works
    assert req("POST", "/logout", jar=jar, base=base) == 200


def test_stream_token_host_allowlist(base=BASE):
    # tokens are plain base64: a crafted token must not let a logged-in user
    # turn the server into a GET proxy for arbitrary URLs
    import base64
    jar = login("alice", "alicepw12345", base)
    bad = base64.urlsafe_b64encode(b"https://evil.example/x.m3u8").decode()
    assert req("GET", f"/stream/{bad}", jar=jar, base=base) == 400
    bad2 = base64.urlsafe_b64encode(b"http://hls.anidb.app/x").decode()  # non-https
    assert req("GET", f"/stream/{bad2}", jar=jar, base=base) == 400


def test_lockout_per_user(base=BASE):
    # must run on a fresh server (in-memory lockout): 5 fails -> 429 for bob,
    # admin unaffected
    codes = [req("POST", "/login", {"username": "bob", "password": "bad"}, base=base)
             for _ in range(6)]
    assert codes[:5] == [401] * 5 and codes[5] == 429, codes
    assert req("POST", "/login", {"username": "admin", "password": "adminpw12345"}, base=base) == 200


# --- helpers ---

import os  # noqa: E402


def _json(method, path, jar=None, base=BASE):
    r = urllib.request.Request(base + path, method=method,
                               headers={"X-Aniweb": "1"})
    opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)) if jar else urllib.request
    try:
        resp = opener.open(r)
    except urllib.error.HTTPError as e:
        raise AssertionError(f"{method} {path} -> {e.code}")
    return json.loads(resp.read().decode())


def _users():
    import sqlite3
    c = sqlite3.connect(os.path.join(os.environ["ANIWEB_DATA"], "aniweb.db"))
    names = [r[0] for r in c.execute("SELECT name FROM users")]
    c.close()
    return names


def _search_count(user):
    import sqlite3
    c = sqlite3.connect(os.path.join(os.environ["ANIWEB_DATA"], "aniweb.db"))
    n = c.execute("SELECT COUNT(*) FROM searches WHERE user=?", (user,)).fetchone()[0]
    c.close()
    return n