bot-ci
297d9279d8
Home omits growth_days; backend auto (or home_growth_days 1/7) picks the window. Response includes days alongside growth_days. Growth榜 link uses the returned days. Non-growth sorts treat omitted/auto as 7.
433 lines
19 KiB
Python
433 lines
19 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"] == 7
|
|
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/theme.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
|
|
heat_payload = json.loads(payload)["data"]
|
|
items = heat_payload["items"]
|
|
assert items and items[0]["aid"] == 102
|
|
# 非增长排序：省略 growth_days 时 days/growth_days 回落为 7，不做 7→1
|
|
assert heat_payload["days"] == heat_payload["growth_days"] == 7
|
|
|
|
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
|
|
|
|
# 首页形态：省略 growth_days → 后端 auto，响应带 days/growth_days
|
|
status, _, payload = running.json("/api/videos?sort=growth&limit=6")
|
|
assert status == 200
|
|
growth = payload["data"]
|
|
assert growth["days"] == growth["growth_days"] == 1
|
|
assert growth["available"] is True
|
|
|
|
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_visit_endpoint_and_stats():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
running = RunningServer(tmp)
|
|
try:
|
|
ua = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/155.0"}
|
|
collector = running.server.context.visits
|
|
|
|
# 上报：返回 204（无响应体），可信对端 + X-Real-IP → 记录真实 IP 与归一化路由
|
|
status, _, body = running.request(
|
|
"/api/visit", method="POST",
|
|
headers={**ua, "Content-Type": "application/json", "X-Real-IP": "203.0.113.7"},
|
|
body=json.dumps({"path": "/video/123"}),
|
|
)
|
|
assert status == 204 and body == b""
|
|
status, _, body = running.request(
|
|
"/api/visit", method="POST",
|
|
headers={**ua, "Content-Type": "application/json", "X-Real-IP": "203.0.113.7"},
|
|
body=json.dumps({"path": "/videos"}),
|
|
)
|
|
assert status == 204 and body == b""
|
|
|
|
assert collector.pending_keys == 2
|
|
routes = {(row["ip"], row["path"]) for row in collector.pending_rows()}
|
|
assert routes == {("203.0.113.7", "/video/:aid"), ("203.0.113.7", "/videos")}
|
|
|
|
# 机器人 UA 被过滤，不计数
|
|
status, _, body = running.request(
|
|
"/api/visit", method="POST",
|
|
headers={"User-Agent": "Googlebot/2.1", "Content-Type": "application/json"},
|
|
body=json.dumps({"path": "/home"}),
|
|
)
|
|
assert status == 204
|
|
assert collector.pending_keys == 2
|
|
|
|
# 超大请求体被拒绝（JSON 错误仍返回）
|
|
status, _, _ = running.request(
|
|
"/api/visit", method="POST",
|
|
headers={**ua, "Content-Type": "application/json"},
|
|
body=json.dumps({"path": "/" + "x" * 4000}),
|
|
)
|
|
assert status == 400
|
|
|
|
# GET 埋点端点 → 405；其他业务 POST 仍 405
|
|
status, _, _ = running.request("/api/visit")
|
|
assert status == 405
|
|
status, _, _ = running.request("/api/videos/101", method="POST")
|
|
assert status == 405
|
|
|
|
# 未 flush 前库里没有行，但 /api/stats 合并内存缓冲可见
|
|
assert running.db.read(
|
|
lambda c: c.execute("SELECT COUNT(*) FROM visit").fetchone()[0]
|
|
) == 0
|
|
status, _, payload = running.json("/api/stats?days=7")
|
|
assert status == 200
|
|
data = payload["data"]
|
|
assert data["pending"] is True
|
|
assert data["today_pv"] == 2
|
|
assert data["today_uv"] == 1
|
|
pages = {page["path"]: page for page in data["top_pages"]}
|
|
assert pages["/video/:aid"]["pv"] == 1
|
|
assert pages["/videos"]["pv"] == 1
|
|
|
|
# 手动 flush 后落库，且 IP 正确
|
|
assert collector.flush() == 2
|
|
assert running.db.read(
|
|
lambda c: c.execute("SELECT COUNT(*) FROM visit").fetchone()[0]
|
|
) == 2
|
|
ips = running.db.read(
|
|
lambda c: [row[0] for row in c.execute("SELECT DISTINCT ip FROM visit")]
|
|
)
|
|
assert ips == ["203.0.113.7"]
|
|
|
|
status, _, payload = running.json("/api/stats?days=7")
|
|
data = payload["data"]
|
|
assert data["pending"] is False
|
|
assert data["today_pv"] == 2
|
|
assert data["range"]["end"] == "2026-10-31"
|
|
finally:
|
|
running.close()
|
|
|
|
|
|
def test_visit_query_path_drains_body():
|
|
"""path 也可来自查询串；请求体必须被读走，避免 keep-alive 连接错位。"""
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
running = RunningServer(tmp)
|
|
try:
|
|
ua = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/155.0"}
|
|
conn = http.client.HTTPConnection("127.0.0.1", running.port, timeout=5)
|
|
conn.request(
|
|
"POST", "/api/visit?path=/about",
|
|
body=json.dumps({"path": "/ignored"}),
|
|
headers={**ua, "Content-Type": "application/json"},
|
|
)
|
|
first = conn.getresponse()
|
|
assert first.status == 204 and first.read() == b""
|
|
# 路径来自查询串而非请求体
|
|
assert running.server.context.visits.pending_rows()[0]["path"] == "/about"
|
|
|
|
# 同一连接继续发请求：若上一次请求体未被读走，这里会解析错位
|
|
conn.request("GET", "/api/health")
|
|
second = conn.getresponse()
|
|
health = json.loads(second.read())
|
|
conn.close()
|
|
assert second.status == 200 and health["ok"] is True
|
|
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, _, body = running.json("/api/stats")
|
|
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/stats", 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_visit_endpoint_and_stats()
|
|
test_visit_query_path_drains_body()
|
|
test_token_auth_and_login()
|
|
test_preview_and_loopback_guards()
|
|
print("OK: Web 只读路由/鉴权/预览/访问统计/安全边界测试全部通过")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main_test()
|