309 lines
13 KiB
Python
309 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""只读 Web 服务测试（离线）: uv run python tests/test_web_server.py"""
|
|
|
|
import http.client
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
from reaction_board.config import ConfigStore # noqa: E402
|
|
from reaction_board.database import Database # noqa: E402
|
|
from reaction_board.web.server import create_server # noqa: E402
|
|
from tests.web_fixtures import build_db, write_config # noqa: E402
|
|
|
|
|
|
class RunningServer:
|
|
def __init__(self, tmp: str, *, token: str = "", preview: bool = False):
|
|
self.config = write_config(tmp, token=token)
|
|
build_db(tmp)
|
|
self.db = Database(Path(tmp) / "app.db")
|
|
self.db.initialize()
|
|
if preview:
|
|
self.db.write(lambda conn: conn.execute(
|
|
"INSERT OR REPLACE INTO kv(k, v) VALUES ('preview:seeded', '1')"
|
|
))
|
|
self.store = ConfigStore(self.config)
|
|
self.server = create_server(
|
|
self.store, self.db, port=0, preview=preview,
|
|
clock=lambda: 1_793_459_000,
|
|
)
|
|
self.port = self.server.server_address[1]
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def request(self, path: str, *, method: str = "GET", headers=None, body=None):
|
|
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=5)
|
|
conn.request(method, path, body=body, headers=headers or {})
|
|
response = conn.getresponse()
|
|
payload = response.read()
|
|
result = response.status, dict(response.getheaders()), payload
|
|
conn.close()
|
|
return result
|
|
|
|
def json(self, path: str, **kwargs):
|
|
status, headers, payload = self.request(path, **kwargs)
|
|
return status, headers, json.loads(payload.decode("utf-8"))
|
|
|
|
def close(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.db.close()
|
|
|
|
|
|
def test_readonly_routes_and_static():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
running = RunningServer(tmp)
|
|
try:
|
|
status, _, health = running.json("/api/health")
|
|
assert status == 200 and health["ok"] is True
|
|
assert health["data"]["preview"] is False
|
|
|
|
status, _, payload = running.json("/api/session")
|
|
assert status == 200
|
|
assert payload["data"]["auth_enabled"] is False
|
|
assert payload["data"]["authenticated"] is True
|
|
status, _, payload = running.json("/api/candidates")
|
|
assert status == 200 and payload["ok"] is True
|
|
status, _, payload = running.json("/api/system")
|
|
assert status == 200 and payload["ok"] is True
|
|
status, _, payload = running.json("/api/about")
|
|
assert status == 200
|
|
assert payload["data"]["schema_version"] == 4
|
|
assert payload["data"]["analytics"]["min_duration_seconds"] == 600
|
|
|
|
status, _, overview = running.json("/api/overview")
|
|
assert status == 200 and overview["ok"] is True
|
|
assert overview["data"]["data_date"] == "2026-10-31"
|
|
|
|
status, headers, payload = running.request("/")
|
|
assert status == 200
|
|
assert "text/html" in headers["Content-Type"]
|
|
assert b"<html" in payload.lower()
|
|
|
|
for asset in ("/static/style.css", "/static/app.js", "/static/vendor/echarts.min.js"):
|
|
status, asset_headers, asset_payload = running.request(asset)
|
|
assert status == 200, asset
|
|
assert asset_payload, asset
|
|
content_type = asset_headers["Content-Type"]
|
|
assert "text/css" in content_type or "javascript" in content_type, asset
|
|
|
|
status, _, payload = running.request("/static/../config.json")
|
|
assert status == 404
|
|
|
|
# 伪静态前端路由：直接访问 /rank、/video/... 返回 index.html
|
|
for route_path in ("/rank", "/videos", "/ups", "/about", "/video/114775988111033", "/up/230020975", "/fans"):
|
|
status, route_headers, route_payload = running.request(route_path)
|
|
assert status == 200, route_path
|
|
assert "text/html" in route_headers["Content-Type"], route_path
|
|
assert b"<html" in route_payload.lower(), route_path
|
|
status, _, _ = running.request("/nope.js")
|
|
assert status == 404
|
|
|
|
status, _, payload = running.request("/api/videos")
|
|
assert status == 200
|
|
items = json.loads(payload)["data"]["items"]
|
|
assert items and items[0]["aid"] == 102
|
|
|
|
status, _, payload = running.json("/api/videos?year=2026")
|
|
assert status == 200
|
|
assert payload["data"]["year"] == 2026
|
|
status, _, payload = running.json("/api/videos?year=1800")
|
|
assert status == 400
|
|
status, _, payload = running.json("/api/ups?year=2026")
|
|
assert status == 200
|
|
assert payload["data"]["year"] == 2026
|
|
|
|
status, _, payload = running.request("/api/ups")
|
|
assert status == 200
|
|
assert json.loads(payload)["data"]["sort"] == "efficiency"
|
|
|
|
status, _, payload = running.request("/api/fans?days=7&recent_days=14")
|
|
assert status == 200
|
|
fans = json.loads(payload)["data"]
|
|
assert fans["recent_days"] == 14
|
|
assert all(
|
|
point["stat_date"] >= "2026-10-18"
|
|
for point in fans["recent_points"]
|
|
)
|
|
assert all("delta" in point for point in fans["recent_points"])
|
|
|
|
status, _, payload = running.request(
|
|
"/api/fans?start=2026-10-29&end=2026-10-31"
|
|
)
|
|
assert status == 200
|
|
data = json.loads(payload)["data"]
|
|
assert data["mode"] == "daily"
|
|
assert data["dates"] == ["2026-10-29", "2026-10-30", "2026-10-31"]
|
|
|
|
running.db.write(lambda conn: conn.execute(
|
|
"INSERT OR REPLACE INTO up_stat"
|
|
"(stat_date, mid, follower, following, captured_at, ok) "
|
|
"VALUES ('2025-01-01', 37754047, 500, NULL, 1, 1)"
|
|
))
|
|
status, _, payload = running.request(
|
|
"/api/fans?start=2025-01-01&end=2026-10-31"
|
|
)
|
|
assert status == 200
|
|
assert json.loads(payload)["data"]["mode"] == "monthly"
|
|
|
|
status, headers, payload = running.request("/api/videos", method="HEAD")
|
|
assert status == 200 and payload == b""
|
|
assert "Content-Length" in headers
|
|
|
|
# 只读：所有业务写方法都被拒绝，且数据库不变
|
|
before = running.db.read(lambda conn: conn.execute(
|
|
"SELECT COUNT(*) FROM candidate"
|
|
).fetchone()[0])
|
|
for method in ("POST", "PUT", "PATCH", "DELETE"):
|
|
status, _, _ = running.request("/api/videos/101", method=method)
|
|
assert status == 405, method
|
|
after = running.db.read(lambda conn: conn.execute(
|
|
"SELECT COUNT(*) FROM candidate"
|
|
).fetchone()[0])
|
|
assert after == before
|
|
|
|
status, _, error = running.json("/api/videos?sort=bad")
|
|
assert status == 400 and error["ok"] is False
|
|
status, _, error = running.json("/api/videos?page=0")
|
|
assert status == 400
|
|
status, _, error = running.json("/api/videos/abc")
|
|
assert status == 400
|
|
status, _, error = running.json("/api/nope")
|
|
assert status == 404
|
|
|
|
status, _, payload = running.request("/api/export.csv?sort=view")
|
|
assert status == 404
|
|
|
|
status, headers, _ = running.request("/api/overview")
|
|
assert headers["X-Content-Type-Options"] == "nosniff"
|
|
assert "default-src 'self'" in headers["Content-Security-Policy"]
|
|
assert headers["X-Frame-Options"] == "DENY"
|
|
finally:
|
|
running.close()
|
|
|
|
|
|
def test_token_auth_and_login():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
running = RunningServer(tmp, token="test-token")
|
|
try:
|
|
# 公共页面与公共 API 访客可见
|
|
status, _, body = running.json("/api/overview")
|
|
assert status == 200 and body["ok"] is True
|
|
status, _, body = running.json("/api/videos")
|
|
assert status == 200 and body["ok"] is True
|
|
status, _, body = running.json("/api/fans?days=7")
|
|
assert status == 200 and body["ok"] is True
|
|
status, _, payload = running.request("/")
|
|
assert status == 200 and b"<html" in payload.lower()
|
|
status, _, payload = running.request("/rank")
|
|
assert status == 200 and b"<html" in payload.lower()
|
|
|
|
# 候选与系统需要登录
|
|
status, _, body = running.json("/api/candidates")
|
|
assert status == 401 and body["error"]["code"] == "unauthorized"
|
|
status, _, body = running.json("/api/system")
|
|
assert status == 401
|
|
status, headers, _ = running.request("/candidates")
|
|
assert status == 303 and headers["Location"] == "/login?next=/candidates"
|
|
status, headers, _ = running.request("/system")
|
|
assert status == 303 and headers["Location"] == "/login?next=/system"
|
|
|
|
status, _, body = running.json("/api/session")
|
|
assert status == 200
|
|
assert body["data"]["auth_enabled"] is True
|
|
assert body["data"]["authenticated"] is False
|
|
|
|
status, _, payload = running.request("/login?next=/candidates")
|
|
assert status == 200 and b"web.auth_token" in payload
|
|
assert b"action=\"/login?next=/candidates\"" in payload
|
|
|
|
status, _, _ = running.request(
|
|
"/login",
|
|
method="POST",
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
body="token=wrong",
|
|
)
|
|
assert status == 401
|
|
|
|
status, headers, _ = running.request(
|
|
"/login?next=/candidates",
|
|
method="POST",
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
body="token=test-token",
|
|
)
|
|
assert status == 303 and headers["Location"] == "/candidates"
|
|
cookie = headers["Set-Cookie"].split(";", 1)[0]
|
|
assert cookie.startswith("rb_session=")
|
|
|
|
status, _, body = running.json("/api/candidates", headers={"Cookie": cookie})
|
|
assert status == 200 and body["ok"] is True
|
|
status, _, body = running.json("/api/system", headers={"Cookie": cookie})
|
|
assert status == 200 and body["ok"] is True
|
|
status, _, body = running.json(
|
|
"/api/candidates", headers={"Authorization": "Bearer test-token"}
|
|
)
|
|
assert status == 200
|
|
|
|
tampered = cookie[:-1] + ("0" if cookie[-1] != "0" else "1")
|
|
status, _, _ = running.json("/api/candidates", headers={"Cookie": tampered})
|
|
assert status == 401
|
|
status, _, body = running.json("/api/overview", headers={"Cookie": tampered})
|
|
assert status == 200
|
|
|
|
status, _, _ = running.json("/api/health")
|
|
assert status == 200
|
|
finally:
|
|
running.close()
|
|
|
|
|
|
def test_preview_and_loopback_guards():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
config = write_config(tmp)
|
|
build_db(tmp)
|
|
db = Database(Path(tmp) / "app.db")
|
|
db.initialize()
|
|
store = ConfigStore(config)
|
|
try:
|
|
try:
|
|
create_server(store, db, host="0.0.0.0", port=0)
|
|
except ValueError as exc:
|
|
assert "web.auth_token" in str(exc)
|
|
else:
|
|
raise AssertionError("非回环地址且无 token 应拒绝启动")
|
|
|
|
try:
|
|
create_server(store, db, port=0, preview=True)
|
|
except ValueError as exc:
|
|
assert "preview:seeded" in str(exc)
|
|
else:
|
|
raise AssertionError("非预览库不应允许 --preview")
|
|
|
|
db.write(lambda conn: conn.execute(
|
|
"INSERT OR REPLACE INTO kv(k, v) VALUES ('preview:seeded', '1')"
|
|
))
|
|
server = create_server(store, db, port=0, preview=True)
|
|
try:
|
|
assert server.context.preview is True
|
|
finally:
|
|
server.server_close()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def main_test():
|
|
test_readonly_routes_and_static()
|
|
test_token_auth_and_login()
|
|
test_preview_and_loopback_guards()
|
|
print("OK: Web 只读路由/鉴权/预览/安全边界测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main_test()
|