361 lines
16 KiB
Python
361 lines
16 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:
|
||
# 读取错误响应体本身也可能超时（尤其 302 无体/远端慢），不能让它把巡检整轮打断
|
||
try:
|
||
body = e.read().decode("utf-8", "replace")
|
||
except Exception: # noqa: BLE001
|
||
body = ""
|
||
return Resp(e.code, dict(e.headers), body)
|
||
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 create_vote(c, title, vtype="single", options=None, fields=None, action="publish"):
|
||
"""提交一次投票创建（POST /user/votes）。
|
||
|
||
后端对创建动作一律按 draft=true 解析（草稿可随时暂存、跳过审核），
|
||
action=publish 时再走 validateForPublish + 内容审核；因此这里两种动作都可直接测。
|
||
"""
|
||
page = c.get("/user/votes/new")
|
||
data = {
|
||
"title": title, "description": "", "visibility": "public",
|
||
"voter_scope": "anyone", "show_result": "after_vote",
|
||
"type": vtype, "max_choices": "1", "allow_custom": "",
|
||
"start_at": "", "end_at": "", "quota_limit": "0",
|
||
"options": json.dumps(options or [], ensure_ascii=False),
|
||
"form_fields": json.dumps(fields or [], ensure_ascii=False),
|
||
"action": action, "_csrf": csrf_of(page.body),
|
||
}
|
||
return c.post("/user/votes", data)
|
||
|
||
|
||
def check_user_assets(owner):
|
||
"""登录态的接口/导出，以及停用页的「身份占位」说明。
|
||
|
||
停用页说明用**结构断言**（.uk-text-muted 段落非空），不写死文案 —— 站点语言随会话变化。
|
||
"""
|
||
r = owner.get("/api/v1/user")
|
||
step("登录态 /api/v1/user 返回 JSON", r.status == 200 and r.body.lstrip().startswith("{"),
|
||
"HTTP %s" % r.status)
|
||
step("登录态 /explore", owner.get("/explore").status == 200)
|
||
|
||
csrf = csrf_of(owner.get("/user/profile").body)
|
||
r = owner.post("/user/profile/export-json", {"_csrf": csrf})
|
||
step("导出 JSON", r.status == 200 and r.body.lstrip().startswith(("{", "[")),
|
||
"HTTP %s" % r.status)
|
||
r = owner.post("/user/profile/export", {"_csrf": csrf})
|
||
step("导出 Excel", r.status == 200, "HTTP %s %s" % (r.status, r.h.get("content-type", "")))
|
||
|
||
r = owner.get("/user/profile/deactivate")
|
||
note = re.search(r'class="uk-text-small uk-text-muted">(.*?)</span>', r.body, re.S)
|
||
step("停用页含身份占位说明", r.status == 200 and bool(note) and note.group(1).strip() != "")
|
||
|
||
|
||
def check_vote_flow(owner, base):
|
||
"""投票全链路：建票发布 -> 管理列表 -> 公开页 -> 匿名提交 -> 导出计数 -> 草稿 -> 表单型。"""
|
||
ts = time.strftime("%m%d%H%M%S")
|
||
|
||
title = "巡检单选投票%s" % ts
|
||
r = create_vote(owner, title, "single", options=["选项甲", "选项乙", "选项丙"])
|
||
step("创建单选投票并发布", r.status == 302 and "questions" in r.loc,
|
||
"HTTP %s -> %s %s" % (r.status, r.loc, r.alert()))
|
||
|
||
q = owner.get("/user/questions")
|
||
step("投票出现在管理列表", title in q.body)
|
||
mt = re.search(r"copyShareURL\('([^']+)'", q.body)
|
||
mv = re.search(r"/user/votes/(\d+)/edit", q.body)
|
||
token, vid = (mt.group(1) if mt else None), (mv.group(1) if mv else None)
|
||
step("取得分享 token / 投票 ID", bool(token and vid), "%s / %s" % (token, vid))
|
||
if not (token and vid):
|
||
return
|
||
|
||
anon = Client(base)
|
||
pv = anon.get("/vote/%s" % token)
|
||
step("公开投票页渲染", pv.status == 200 and title in pv.body, "HTTP %s" % pv.status)
|
||
mo = re.search(r'name="option_id" value="(\d+)"', pv.body)
|
||
step("公开页含可选项", bool(mo))
|
||
if not mo:
|
||
return
|
||
|
||
sub = anon.post("/vote/%s/submit" % token, {
|
||
"option_id": mo.group(1), "g-recaptcha-response": captcha(anon),
|
||
"custom_option": "", "_csrf": csrf_of(pv.body),
|
||
})
|
||
step("匿名提交投票", sub.status == 302, "HTTP %s -> %s %s" % (sub.status, sub.loc, sub.alert()))
|
||
|
||
# 导出仅作只读校验：单选导出的 CSV 每行是 <选项,票数>，末行为 <总参与,N>。
|
||
# 只认「总参与」这一行，避免把各选项票数一起求和导致 2 倍计数。
|
||
exp = owner.get("/user/votes/%s/export" % vid)
|
||
mtot = re.search(r"总参与,(\d+)", exp.body)
|
||
total = int(mtot.group(1)) if mtot else -1
|
||
step("导出显示总参与 = 1", exp.status == 200 and total == 1,
|
||
"HTTP %s 总参与=%s" % (exp.status, total))
|
||
|
||
dr = create_vote(owner, "巡检草稿%s" % ts, "single", options=["甲", "乙"], action="save")
|
||
step("保存草稿跳编辑页(?saved=1)",
|
||
dr.status == 302 and "/user/votes/" in dr.loc and "saved=1" in dr.loc,
|
||
"HTTP %s -> %s" % (dr.status, dr.loc))
|
||
|
||
fr = create_vote(owner, "巡检表单投票%s" % ts, "form", fields=[
|
||
{"field_type": "text", "label": "昵称", "required": True, "max_length": 50},
|
||
{"field_type": "textarea", "label": "建议", "required": False, "max_length": 200},
|
||
])
|
||
step("创建表单型投票并发布", fr.status == 302 and "questions" in fr.loc,
|
||
"HTTP %s -> %s %s" % (fr.status, fr.loc, fr.alert()))
|
||
|
||
|
||
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", "agree_rules": "1",
|
||
"_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", "agree_rules": "1",
|
||
"_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)
|
||
check_user_assets(owner)
|
||
check_vote_flow(owner, base)
|
||
|
||
# 匿名提问：无邮箱时成功后直接 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", "agree_rules": "1",
|
||
"_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__":
|
||
try:
|
||
sys.exit(main())
|
||
except KeyboardInterrupt:
|
||
raise
|
||
except Exception as e: # noqa: BLE001
|
||
# 远端抖动等意外中断时，仍把已完成项的结论打出来，避免只剩一段 traceback
|
||
print("\n巡检意外中断：%s: %s" % (type(e).__name__, e))
|
||
for d, ok, ex in RESULTS:
|
||
print("%s %s%s" % ("PASS" if ok else "FAIL", d, (" (%s)" % ex) if ex else ""))
|
||
sys.exit(2)
|