jozamudi
d90b1af8d6
- app.js: pause/pagehide listeners passed the Event into savePos(floor), making pos NaN -> serialized as null -> every pause/unload clobbered the saved position. Wrap in arrows; pagehide now posts keepalive so the final save survives page unload - app.js: play() re-fetches history on a miss (boot fetch swallows errors) and only bumps an existing row instead of posting pos:0 at start-of-play - app.py: POST /api/history requires slug_id (was stored as "None"), validates/clamps pos; upsert keeps the old payload when the incoming pos is null so stale clients can't corrupt a row - tests: +3 API tests (slug required, pos clamping, null-pos keeps saved), +1 end-to-end browser test (resume ~400s with a failed boot fetch); lockout test now uses a throwaway username so it can't lock out users the browser tests log in with; hardened the flaky autoplay test
445 lines
19 KiB
Python
445 lines
19 KiB
Python
"""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_requires_slug(base=BASE):
|
|
jar = login("admin", "adminpw12345", base)
|
|
# a missing slug_id used to be stored as a "None" ghost row
|
|
assert req("POST", "/api/history", {"ep_id": 991}, jar=jar, base=base) == 422
|
|
|
|
|
|
def test_history_clamps_pos(base=BASE):
|
|
jar = login("admin", "adminpw12345", base)
|
|
e = {"ep_id": 992, "slug_id": "junk-a", "title": "Junk", "season_id": 1,
|
|
"ep": "1", "lang": "eng", "pos": -50, "dur": 100}
|
|
assert req("POST", "/api/history", e, jar=jar, base=base) == 200
|
|
got = [x for x in _json("GET", "/api/history?all=1", jar=jar, base=base)
|
|
if x["slug_id"] == "junk-a"]
|
|
assert got and got[0]["pos"] == 0.0, got
|
|
e["pos"] = 999999
|
|
assert req("POST", "/api/history", e, jar=jar, base=base) == 200
|
|
got = [x for x in _json("GET", "/api/history?all=1", jar=jar, base=base)
|
|
if x["slug_id"] == "junk-a"]
|
|
assert got and got[0]["pos"] == 100.0, got
|
|
# non-numeric pos is rejected, not stored as junk
|
|
e["pos"] = "somenonsense"
|
|
assert req("POST", "/api/history", e, jar=jar, base=base) == 422
|
|
|
|
|
|
def test_history_null_pos_keeps_saved(base=BASE):
|
|
jar = login("dave", "davepw12345", base)
|
|
good = {"ep_id": 993, "slug_id": "keep-a", "title": "Keep", "season_id": 1,
|
|
"ep": "1", "lang": "eng", "pos": 400, "dur": 1500}
|
|
assert req("POST", "/api/history", good, jar=jar, base=base) == 200
|
|
# a stale client (old NaN-pos bug) re-POSTs pos null: the upsert must
|
|
# keep the row's saved position instead of clobbering it to 0
|
|
bad = dict(good)
|
|
bad["pos"] = None
|
|
assert req("POST", "/api/history", bad, jar=jar, base=base) == 200
|
|
got = [x for x in _json("GET", "/api/history?all=1", jar=jar, base=base)
|
|
if x["slug_id"] == "keep-a"]
|
|
assert got and got[0]["pos"] == 400, got
|
|
|
|
|
|
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_stats(base=BASE):
|
|
import sqlite3, time
|
|
jar = login("erin", "erinpw12345", base)
|
|
# fresh user: zeroed stats
|
|
d = _json("GET", "/api/stats?days=7", jar=jar, base=base)
|
|
assert d["watch_minutes"] == 0 and d["episodes"] == {"started": 0, "completed": 0}
|
|
# synthetic viewing: ep 1 = 3 position events, pos deltas of 80s each
|
|
# (160s watched); ep 2 watched to 1190s of a 1200s dur (completed);
|
|
# one search. All an hour ago, jpn.
|
|
c = sqlite3.connect(os.path.join(os.environ["ANIWEB_DATA"], "aniweb.db"))
|
|
t0 = time.time() - 3600
|
|
e = {"slug_id": "stat-a", "title": "Stat Show", "lang": "jpn",
|
|
"season_id": 1, "dur": 1200, "total": 24, "poster": "https://p/1.jpg"}
|
|
for i, pos in enumerate((0.0, 80.0, 160.0)):
|
|
c.execute("INSERT INTO events (ts, user, type, payload) VALUES (?,?,?,?)",
|
|
(t0 + 10 * i, "erin", "position",
|
|
json.dumps({**e, "ep_id": 900, "ep": "1", "pos": pos, "t": t0 + 10 * i})))
|
|
for i, pos in enumerate((1100.0, 1190.0)):
|
|
c.execute("INSERT INTO events (ts, user, type, payload) VALUES (?,?,?,?)",
|
|
(t0 + 30 + 10 * i, "erin", "position",
|
|
json.dumps({**e, "ep_id": 901, "ep": "2", "pos": pos, "t": t0 + 30 + 10 * i})))
|
|
c.execute("INSERT INTO events (ts, user, type, payload) VALUES (?,?,?,?)",
|
|
(t0, "erin", "search", json.dumps({"q": "stat anime"})))
|
|
c.commit()
|
|
c.close()
|
|
d = _json("GET", "/api/stats?days=7", jar=jar, base=base)
|
|
assert d["watch_minutes"] == 4, d # round((160+90)/60)
|
|
assert d["episodes"] == {"started": 2, "completed": 1}
|
|
assert d["days_active"] == 1 and d["best_streak"] == 1
|
|
assert d["on_hold"] == 0
|
|
assert sum(d["by_weekday"]) == 250 # seconds, Mon..Sun
|
|
assert sum(d["by_day"].values()) == 250 and all(isinstance(v, int) and v > 0 for v in d["by_day"].values())
|
|
a0 = d["top_anime"][0]
|
|
assert a0["title"] == "Stat Show" and a0["status"] == "ongoing", a0
|
|
assert a0["cur_ep"] == "2" and a0["total"] == 24 and a0["poster"] == "https://p/1.jpg"
|
|
assert a0["sub"] == 250 and a0["dub"] == 0 and a0["pace"] == 2.0 and a0["rewatches"] == 0
|
|
assert d["top_searches"] == ["stat anime"]
|
|
assert d["sub_dub"]["jpn"] >= 1 and d["sub_dub"]["eng"] == 0
|
|
# slug filter scopes to one anime
|
|
ds = _json("GET", "/api/stats?days=7&slug=stat-a", jar=jar, base=base)
|
|
assert len(ds["top_anime"]) == 1 and ds["watch_minutes"] == 4
|
|
assert sum(ds["by_day"].values()) == 250
|
|
dn = _json("GET", "/api/stats?days=7&slug=none", jar=jar, base=base)
|
|
assert dn["top_anime"] == [] and dn["watch_minutes"] == 0 and dn["by_day"] == {}
|
|
# per-user isolation: alice never watched stat-a
|
|
a = login("alice", "alicepw12345", base)
|
|
da = _json("GET", "/api/stats?days=7", jar=a, base=base)
|
|
assert all(x["title"] != "Stat Show" for x in da["top_anime"])
|
|
# days window is validated
|
|
assert req("GET", "/api/stats?days=0", jar=jar, 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):
|
|
# a throwaway name: the lockout is per-username and in-memory, and bob is
|
|
# also used by the browser tests (a locked-out bob breaks their logins)
|
|
codes = [req("POST", "/login", {"username": "lockout-ghost", "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
|
|
|
|
|
|
def test_parse_intros():
|
|
from server.upstream import _parse_intros
|
|
out = _parse_intros([
|
|
{"intro": {"start": 95, "length": 60}, "title": "Ryomen Sukuna"},
|
|
{"intro": {"start": 0, "length": 0}}, # MAL "no intro" marker
|
|
{"intro": None, "title": "Named, no timing"},
|
|
{},
|
|
])
|
|
assert out == [
|
|
{"title": "Ryomen Sukuna", "start": 95, "length": 60},
|
|
{"title": None, "start": 0, "length": 0},
|
|
{"title": "Named, no timing"},
|
|
{"title": None},
|
|
]
|
|
# setupIntro relies on `t.length` being absent (falsy) when no timing
|
|
assert "start" not in out[2]
|
|
|
|
|
|
def test_intros_parses_jikan_shapes():
|
|
# _intros against a stubbed HTTP layer: Jikan v4 returns the episode
|
|
# list directly under 'data' (not {'episodes': [...]}) — a regression
|
|
# here would silently disable the skip button whenever Jikan is up.
|
|
import server.upstream as u
|
|
|
|
class R:
|
|
def json(self):
|
|
return {"data": [
|
|
{"title": "Episode 1", "intro": {"start": 111, "length": 77}},
|
|
{"title": "Episode 2", "intro": {"start": 0, "length": 0}},
|
|
]}
|
|
|
|
u._CACHE.clear()
|
|
real_get = u._get
|
|
u._get = lambda url: R() # stub the network
|
|
assert u._intros(92573) == [
|
|
{"title": "Episode 1", "start": 111, "length": 77},
|
|
{"title": "Episode 2", "start": 0, "length": 0},
|
|
]
|
|
# wrapped shape still works
|
|
class W:
|
|
def json(self):
|
|
return {"data": {"episodes": [
|
|
{"intro": {"start": 90, "length": 60}},
|
|
]}}
|
|
u._get = lambda url: W()
|
|
assert u._intros(92573)[0]["start"] == 90
|
|
u._get = real_get # restore: later tests need the real _get
|
|
u._CACHE.clear()
|
|
|
|
|
|
def test_fetch_parallel_stitches_range_lanes():
|
|
# fetch_parallel across N byte-range lanes must reassemble byte-identical,
|
|
# and must raise (so the app falls back to the single-request path) when
|
|
# the CDN refuses ranges or HEAD fails.
|
|
import server.upstream as u
|
|
|
|
body = bytes((i * 7 + 3) % 256 for i in range(1024 * 1024)) # 1MiB > 2*lane, forces the lane path
|
|
|
|
class Resp:
|
|
def __init__(self, content, status, clen=None):
|
|
self.content, self.status_code = content, status
|
|
self.headers = {"content-type": "video/mp2t"}
|
|
if clen is not None:
|
|
self.headers["content-length"] = str(clen)
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise RuntimeError(f"http {self.status_code}")
|
|
@property
|
|
def text(self):
|
|
return ""
|
|
|
|
class Ranges:
|
|
def head(self, url, **kw):
|
|
return Resp(b"", 200, len(body))
|
|
def get(self, url, **kw):
|
|
rng = (kw.get("headers") or {}).get("Range")
|
|
if rng:
|
|
a, b = rng[len("bytes="):].split("-")
|
|
return Resp(body[int(a):int(b) + 1], 206)
|
|
return Resp(body, 200)
|
|
|
|
class NoRanges: # 416 on every range -> must raise, app falls back
|
|
def head(self, url, **kw):
|
|
return Resp(b"", 200, len(body))
|
|
def get(self, url, **kw):
|
|
if (kw.get("headers") or {}).get("Range"):
|
|
return Resp(b"", 416)
|
|
return Resp(body, 200)
|
|
|
|
real = u._session
|
|
real_get = u._get
|
|
try:
|
|
u._session = lambda: Ranges()
|
|
out, ct = u.fetch_parallel("http://cdn/x.xls", lanes=4)
|
|
assert out == body and ct == "video/mp2t"
|
|
|
|
u._session = lambda: NoRanges()
|
|
try:
|
|
u.fetch_parallel("http://cdn/x.xls", lanes=4)
|
|
raise AssertionError("should have raised")
|
|
except RuntimeError:
|
|
pass
|
|
finally:
|
|
u._session = real
|
|
u._get = real_get
|