190 lines
7.7 KiB
Python
190 lines
7.7 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"})
|
|
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"})
|
|
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 json.load(open(_users_file()))
|
|
|
|
|
|
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)
|
|
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_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)
|
|
assert req("POST", "/account/delete", jar=jar, base=base) == 200
|
|
assert "dave" not in json.load(open(_users_file()))
|
|
# 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_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 ---
|
|
|
|
def _json(method, path, jar=None, base=BASE):
|
|
r = urllib.request.Request(base + path, method=method)
|
|
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_file():
|
|
import os
|
|
return os.path.join(os.environ["ANIWEB_DATA"], "users.json")
|