fix some bugs.
This commit is contained in:
Binary file not shown.
+114
-3
@@ -3,7 +3,8 @@
|
||||
"""TamaBox 部署后冒烟巡检(端到端)。
|
||||
|
||||
用途:对已部署环境跑一遍核心链路,用于发版/重新部署后的回归。
|
||||
覆盖:注册 -> 登录 -> 个人页/提问列表/投票面板 -> 匿名提问 -> 停用
|
||||
覆盖:注册 -> 登录 -> 个人页/提问列表/投票面板 -> 投票全链路(建票发布/公开页/匿名提交/
|
||||
结果导出计数/草稿/表单型)-> 匿名提问 -> 停用
|
||||
-> /deactivated 成功页 -> **同邮箱重注册必须被明确拒绝并提示联系站长**
|
||||
-> 公开页只读巡检 -> /deactivated 四语言渲染。
|
||||
|
||||
@@ -74,7 +75,12 @@ class Client:
|
||||
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"))
|
||||
# 读取错误响应体本身也可能超时(尤其 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))
|
||||
|
||||
@@ -138,6 +144,100 @@ def check_public_pages(base):
|
||||
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 = []
|
||||
@@ -203,6 +303,8 @@ def main():
|
||||
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)
|
||||
@@ -246,4 +348,13 @@ def main():
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(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)
|
||||
|
||||
Reference in New Issue
Block a user