250 lines
10 KiB
Python
250 lines
10 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""TamaBox 部署后冒烟巡检（端到端）。
|
||
|
||
用途：对已部署环境跑一遍核心链路，用于发版/重新部署后的回归。
|
||
覆盖：注册 -> 登录 -> 个人页/提问列表/投票面板 -> 匿名提问 -> 停用
|
||
-> /deactivated 成功页 -> **同邮箱重注册必须被明确拒绝并提示联系站长**
|
||
-> 公开页只读巡检 -> /deactivated 四语言渲染。
|
||
|
||
语义约定（2026-09-15 起）：停用 = 清空内容 + 保留用户行作「身份占位」，
|
||
故邮箱/域名继续被占用，重注册应得到「账号已停用，请找站长」的提示而不是 302 或 500。
|
||
|
||
用法：
|
||
python dev/devcheck-e2e.py [--base https://box-dev-api.337845818.best]
|
||
|
||
注意：
|
||
- 会在目标环境**真实创建账号并发送提问**（使用 tbe2e<时间戳>@example.com 之类的可识别测试邮箱），
|
||
仅在测试/开发环境运行。
|
||
- 每次完整运行都会**留下一个已停用的身份占位**（邮箱/域名被占用）。清理方式：
|
||
`TamaBox ban --email <该邮箱> --yes`（连行物理删除）或 `TamaBox restore --email <该邮箱> --yes`（恢复账号）。
|
||
- 依赖数学验证码（conf.Recaptcha.CaptchaType == "math"）；若站点改用 recaptcha/turnstile 则无法自动通过。
|
||
"""
|
||
import argparse
|
||
import http.cookiejar
|
||
import json
|
||
import re
|
||
import sys
|
||
import time
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
|
||
|
||
RESULTS = []
|
||
|
||
|
||
class NoRedirect(urllib.request.HTTPRedirectHandler):
|
||
def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: D102
|
||
return None
|
||
|
||
|
||
class Resp:
|
||
def __init__(self, status, headers, body):
|
||
self.status = status
|
||
self.h = {k.lower(): v for k, v in headers.items()}
|
||
self.body = body
|
||
|
||
@property
|
||
def loc(self):
|
||
return self.h.get("location", "")
|
||
|
||
def alert(self):
|
||
m = re.search(r'class="uk-alert[^"]*"[^>]*>(.*?)</div>', self.body, re.S)
|
||
return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", m.group(0))).strip()[:200] if m else ""
|
||
|
||
|
||
class Client:
|
||
def __init__(self, base):
|
||
self.base = base
|
||
self.cj = http.cookiejar.CookieJar()
|
||
self.op = urllib.request.build_opener(
|
||
urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())
|
||
|
||
def req(self, method, path, data=None, timeout=90):
|
||
url = path if path.startswith("http") else self.base + path
|
||
headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
|
||
body = urllib.parse.urlencode(data).encode() if data is not None else None
|
||
if body:
|
||
headers["Content-Type"] = "application/x-www-form-urlencoded"
|
||
req = urllib.request.Request(url, data=body, headers=headers, method=method)
|
||
try:
|
||
with self.op.open(req, timeout=timeout) as r:
|
||
return Resp(r.status, dict(r.headers), r.read().decode("utf-8", "replace"))
|
||
except urllib.error.HTTPError as e:
|
||
return Resp(e.code, dict(e.headers), e.read().decode("utf-8", "replace"))
|
||
except Exception as e: # noqa: BLE001
|
||
return Resp(-1, {}, "TRANSPORT_ERROR: %s: %s" % (type(e).__name__, e))
|
||
|
||
def get(self, p, **kw):
|
||
# 只读 GET 重试一次：远端偶发慢请求会产生 TRANSPORT_ERROR，避免巡检出现假失败。
|
||
r = self.req("GET", p, **kw)
|
||
if r.status == -1:
|
||
r = self.req("GET", p, **kw)
|
||
return r
|
||
|
||
def post(self, p, data=None, **kw):
|
||
return self.req("POST", p, data=data, **kw)
|
||
|
||
|
||
def csrf_of(html):
|
||
m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
|
||
return m.group(1) if m else None
|
||
|
||
|
||
def captcha(c):
|
||
"""取一道数学验证码并算出答案，返回 <id>:<answer>。"""
|
||
j = json.loads(c.get("/api/captcha/generate").body)
|
||
pid, q = j["data"]["id"], j["data"]["question"]
|
||
m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
|
||
a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
|
||
ans = {"+": a + b, "-": a - b, "×": a * b}[op]
|
||
return "%s:%d" % (pid, ans)
|
||
|
||
|
||
def step(desc, ok, extra=""):
|
||
RESULTS.append((desc, ok, extra))
|
||
print(" [%s] %s%s" % ("PASS" if ok else "FAIL", desc, (" (%s)" % extra) if extra else ""))
|
||
|
||
|
||
# 停用提示的跨语言特征串（站点默认语言可能是 ja，断言不能只看中文）
|
||
DEACTIVATED_HINTS = ("已停用", "has been deactivated", "は停止されています")
|
||
INTERNAL_ERROR_HINTS = ("サーバー内部エラー", "Internal Server Error", "服务器内部错误")
|
||
|
||
|
||
def is_deactivated_rejected(r):
|
||
"""停用后的重注册应被明确拒绝：HTTP 200 + 提示「账号已停用」，且不得是 500 兜底。"""
|
||
if r.status != 200:
|
||
return False
|
||
if any(h in r.body for h in INTERNAL_ERROR_HINTS):
|
||
return False
|
||
return any(h in r.body for h in DEACTIVATED_HINTS)
|
||
|
||
|
||
def check_public_pages(base):
|
||
"""公开页只读巡检：期望全部 200（/user/votes 之类的登录页除外）。"""
|
||
paths = ["/", "/explore", "/register", "/login", "/forgot-password",
|
||
"/sponsor", "/change-logs", "/about-me", "/robots.txt",
|
||
"/sitemap.xml", "/llms.txt", "/llms-full.txt", "/favicon.ico",
|
||
"/deactivated", "/api/captcha/generate", "/api/v1/site-stats"]
|
||
c = Client(base)
|
||
bad = []
|
||
for p in paths:
|
||
r = c.get(p, timeout=40)
|
||
if r.status != 200:
|
||
bad.append("%s=%s" % (p, r.status))
|
||
step("公开页只读巡检（%d 个端点）" % len(paths), not bad, ", ".join(bad))
|
||
|
||
|
||
def check_deactivated_i18n(base):
|
||
"""停用成功页四语言渲染（用语言无关的 ratio:3 标记 + 标题非空）。"""
|
||
bad = []
|
||
for lang in ["zh-CN", "zh-TW", "en", "ja"]:
|
||
c = Client(base)
|
||
c.get("/i18n/%s?to=/deactivated" % lang, timeout=40) # 302，语言写入 session
|
||
r = c.get("/deactivated", timeout=40) # 再请求一次才渲染新语言
|
||
title = re.search(r'uk-card-body uk-text-center".*?<h3[^>]*>([^<]*)</h3>',
|
||
r.body, re.S)
|
||
if not title or not title.group(1).strip():
|
||
bad.append(lang)
|
||
step("停用成功页四语言渲染", not bad, "缺失: " + ",".join(bad) if bad else "")
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--base", default="https://box-dev-api.337845818.best")
|
||
ap.add_argument("--reuse-email", default="",
|
||
help="只跑一项：用该既有邮箱重新注册（用于复测停用后能否重注册）")
|
||
args = ap.parse_args()
|
||
base = args.base.rstrip("/")
|
||
pwd = "TbCheck!2026"
|
||
|
||
if args.reuse_email:
|
||
domain = args.reuse_email.split("@")[0]
|
||
c = Client(base)
|
||
r = c.get("/register")
|
||
r = c.post("/register", {
|
||
"email": args.reuse_email, "domain": domain, "name": "TBCheck",
|
||
"password": pwd, "repeat_password": pwd,
|
||
"g-recaptcha-response": captcha(c), "language": "zh-CN",
|
||
"_csrf": csrf_of(r.body),
|
||
})
|
||
step("以既有邮箱重新注册（已停用则应被明确拒绝并提示找站长）", is_deactivated_rejected(r),
|
||
"HTTP %s %s" % (r.status, r.alert()))
|
||
return 0 if all(x[1] for x in RESULTS) else 1
|
||
|
||
ts = time.strftime("%m%d%H%M%S")
|
||
email, domain = "tbe2e%s@example.com" % ts, "tbe2e%s" % ts
|
||
print("== 新账号：%s / %s ==" % (email, domain))
|
||
|
||
pub = Client(base)
|
||
r = pub.get("/register")
|
||
reg = pub.post("/register", {
|
||
"email": email, "domain": domain, "name": "TB%s" % ts,
|
||
"password": pwd, "repeat_password": pwd,
|
||
"g-recaptcha-response": captcha(pub), "language": "zh-CN",
|
||
"_csrf": csrf_of(r.body),
|
||
})
|
||
step("注册", reg.status == 302, "HTTP %s -> %s %s" % (reg.status, reg.loc, reg.alert()))
|
||
|
||
owner = Client(base)
|
||
r = owner.get("/login")
|
||
r = owner.post("/login", {
|
||
"email": email, "password": pwd,
|
||
"g-recaptcha-response": captcha(owner), "_csrf": csrf_of(r.body),
|
||
})
|
||
step("登录", r.status == 302, "HTTP %s -> %s" % (r.status, r.loc))
|
||
|
||
step("登录态 /user/profile", owner.get("/user/profile").status == 200)
|
||
step("登录态 /user/questions", owner.get("/user/questions").status == 200)
|
||
r = owner.get("/user/votes")
|
||
step("/user/votes 302 -> /user/questions#votes",
|
||
r.status == 302 and "questions" in r.loc, "HTTP %s -> %s" % (r.status, r.loc))
|
||
step("新建投票页", owner.get("/user/votes/new").status == 200)
|
||
|
||
# 匿名提问：无邮箱时成功后直接 200 渲染提问成功页（不重定向），故只断言提问详情可达
|
||
r = pub.get("/_/%s" % domain)
|
||
step("打开提问箱", r.status == 200)
|
||
r = pub.post("/_/%s" % domain, {
|
||
"content": "自动化巡检测试提问（可忽略）",
|
||
"g-recaptcha-response": captcha(pub),
|
||
"receive_reply_email": "", "private": "", "ask_password": "",
|
||
"start_chat_val": "", "_csrf": csrf_of(r.body),
|
||
})
|
||
step("提交提问", r.status in (200, 302), "HTTP %s -> %s" % (r.status, r.loc))
|
||
|
||
r = owner.get("/user/profile/deactivate")
|
||
step("打开停用页", r.status == 200)
|
||
r = owner.post("/user/profile/deactivate", {"_csrf": csrf_of(r.body)})
|
||
step("停用后跳 /deactivated", r.status == 302 and "deactivated" in r.loc,
|
||
"HTTP %s -> %s" % (r.status, r.loc))
|
||
r = owner.get("/deactivated")
|
||
step("停用成功页渲染", r.status == 200 and "ratio: 3" in r.body, "HTTP %s" % r.status)
|
||
|
||
pub2 = Client(base)
|
||
r = pub2.get("/register")
|
||
r = pub2.post("/register", {
|
||
"email": email, "domain": domain, "name": "TB%sb" % ts,
|
||
"password": pwd, "repeat_password": pwd,
|
||
"g-recaptcha-response": captcha(pub2), "language": "zh-CN",
|
||
"_csrf": csrf_of(r.body),
|
||
})
|
||
step("停用后同邮箱重注册被明确拒绝并提示找站长（核心回归）", is_deactivated_rejected(r),
|
||
"HTTP %s %s" % (r.status, r.alert()))
|
||
|
||
check_public_pages(base)
|
||
check_deactivated_i18n(base)
|
||
|
||
print("\n===== 汇总 =====")
|
||
for d, ok, ex in RESULTS:
|
||
print("%s %s%s" % ("PASS" if ok else "FAIL", d, (" (%s)" % ex) if ex else ""))
|
||
n = sum(1 for x in RESULTS if not x[1])
|
||
print("\n共 %d 项，失败 %d 项" % (len(RESULTS), n))
|
||
return 1 if n else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|