Files
xiaotutongxue/scripts/e2e-m2-join.py
T
lovife 30c3714d35 test(scripts): 36 个 E2E 与 6 个机械校验工具(1548 条断言)+ 一键交付自检
新增:check-api-parity(契约 121 接口 ↔ 控制器,自带「必须恰好 121」自检)、
check-entity-columns(基类要求的列 vs 迁移建表列,F-35 那个坑踩了四次之后上的工具)、
check-manual(手册引用的错误码与提示文案对得上代码)、check-test-profile(test profile 真起一次)、
e2e-prodguard、load-test(四场景、按端点分别统计)、verify-all(一条命令跑完交付自检)、
dev-seed-audit(界面验收用的演示数据),以及 M1~M9 各轮 E2E。

压测抓到一个真缺陷并修掉(F-92):门禁 IP 限流原来计「所有请求」,于是一个班连同一个
WiFi 时 60 个不同学生同时核验只过了 20 个 → 改为只计失败 + 单独的成功上限,20/60 → 60/60。

度量工具的偏差会伪装成产品缺陷(F-60/F-77/F-97),所以每个脚本都断言「动作自己的结论字段」
(submitted / created / canExport),而不是 HTTP 200 + code 0。

验证:verify-all.py --quick 5/5;--with-e2e 见 docs/09 §8。
2026-09-30 09:02:57 +08:00

407 lines
19 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# -*- coding: utf-8 -*-
"""B1 名单加入申请 E2E(FR-509,《03-接口设计》§8.7)。
它验的是这条通道的**四条边界**(都是刻意的设计,不是"随手加的校验"):
1. **不绕过门禁**:申请期间学生**仍然不能填报** —— 这条最容易被实现成
"提交申请就能填",那样门禁就变成形式主义,整套"凭证件号认人"的根基没了。
2. **已在名册 → 40039 让他回去重新核验**,而不是受理申请(他只是姓名输错了)。
3. **每人每任务最多 2 次**(40037):没有上限的话,一个人能刷出几十条待办。
4. **通过即自动绑定**:学生返回重新核验就能直接进,不用再输一遍姓名与证件号。
跑法(前提:后端 8080 在跑、xt-pg 在跑;脚本自己摆数据、自己清理):
python scripts/e2e-m2-join.py
"""
from __future__ import annotations
import argparse
import importlib.util
import json
import os
import sys
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []
def _load(module_name: str, filename: str):
"""复用既有 E2E 的工具与夹具,不复制第二份(两份必然分叉)。"""
path = os.path.join(ROOT, "scripts", filename)
spec = importlib.util.spec_from_file_location(module_name, path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
R = _load("e2e_m2_roster", "e2e-m2-roster.py")
BASE = "http://localhost:8080"
CREATED_TASK_IDS: list[int] = []
TMP_ROSTER = os.path.join(ROOT, "scripts", "_e2e-join-roster.xlsx")
# 名册里的那个人(老师已导入)
IN_ROSTER_NAME = "曾伊豪"
IN_ROSTER_ID = R.ID_A
# 不在名册里的人(要申请加入)——身份证号用校验位合法的
OUT_NAME = "新来同学"
OUT_ID = R.with_checksum("36102920090101005")
OUT_PHONE = "13900139009"
def ok(name: str) -> None:
PASSED.append(name)
print(f" [OK] {name}")
def fail(name: str, detail: str) -> None:
FAILED.append((name, detail))
print(f" [FAIL] {name}\n {detail}")
def check(name: str, cond: bool, detail: str = "") -> bool:
if cond:
ok(name)
else:
fail(name, detail or "断言不成立")
return cond
def sql(stmt: str) -> str:
okk, out = R.sql(stmt)
if not okk:
fail("SQL 执行失败", f"{stmt[:80]} → {out}")
return out
def body_of(method: str, path: str, token: str | None = None, payload: dict | None = None) -> dict:
_, b = R.call(method, f"{BASE}{path}", token, payload)
return b
def ensure_student(uid: int) -> str:
"""
准备一个"有 openid 的学生",返回他的 token。
为什么要自己建账号:这条通道按 **openid** 限次与去重(契约 §8.7.1),
而"从微信进来的人才有 openid"是产品的真实前提。用 SQL 直接造出带 open_id 的账号,
是把"微信授权"这一步(需要真公众号,E2E 走不了)替换成等价的前置数据 ——
**其余全走真实接口**。`ON CONFLICT` 让脚本可以重复跑。
"""
openid = f"e2e-join-{uid}"
R.sql(f"""
INSERT INTO sys_user (id, tenant_id, username, real_name, open_id, user_type, status)
VALUES ({uid}, 1, 'join{uid}', '加入申请学生{uid}', '{openid}', 4, 1)
ON CONFLICT (id) DO UPDATE SET open_id = EXCLUDED.open_id, status = 1
""")
cfg = R.read_dev_config()
return R.mint_mp_token(cfg["jwt_secret"], cfg["issuer"], uid, 1, ["STUDENT"], 4)
def main() -> int:
ap = argparse.ArgumentParser(description="B1 名单加入申请 E2E")
ap.parse_args()
print("=" * 74)
print(" B1 名单加入申请 E2E(FR-509)")
print("=" * 74)
_, adm = R.login(BASE, "admin", "Admin@123", "demo")
if adm.get("code") != 0:
print(f" ✗ 管理员登录失败:{adm}")
return 2
at = adm["data"]["token"]
cfg = R.read_dev_config()
# ── 0. 摆数据:一个收集中的任务 + 名册里一个人 ──
print("\n[0] 摆夹具:任务 + 名册(一个人)+ 一个不在名册的学生")
suffix = os.urandom(3).hex()
_, b = R.call("POST", f"{BASE}/api/v1/tasks", at, {
"taskName": f"E2E加入申请 {suffix}", "batchNo": "202609", "majorName": "数控加工",
"collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31"
})
if b.get("code") != 0:
print(f" ✗ 建任务失败:{b}")
return 2
task_id = b["data"]["id"]
CREATED_TASK_IDS.append(task_id)
_, b = R.call("GET", f"{BASE}/api/v1/orgs/tree", at)
classes = []
def walk(nodes):
for n in nodes or []:
if n.get("orgType") == 3:
classes.append(n)
walk(n.get("children"))
walk(b.get("data") or [])
if not classes:
print(" ✗ demo 租户里没有班级")
return 2
org = classes[0]
_, b = R.call("PUT", f"{BASE}/api/v1/tasks/{task_id}/orgs", at, {
"orgs": [{"orgId": org["id"], "expectedCount": 2}]
})
if b.get("code") != 0:
print(f" ✗ 配班级失败:{b}")
return 2
_, b = R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/materials/apply-template", at,
{"templateCode": "TPL_SCHOOL_ROLL"})
if b.get("code") != 0:
print(f" ✗ 应用材料模板失败:{b}")
return 2
R.build_fixture(TMP_ROSTER)
body = R.upload_roster(BASE, at, task_id, TMP_ROSTER,
extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]})
check("名册导入成功", body.get("code") == 0, json.dumps(body, ensure_ascii=False)[:200])
_, b = R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/publish", at, {"force": True})
check("任务发布成功(收集中)", b.get("code") == 0,
json.dumps(b, ensure_ascii=False)[:200])
out_tok = ensure_student(993301)
# ── 1. 门禁失败:不在名册里的人进不去 ──
print("\n[1] 不在名册的人:门禁必须拦住(这是产品根基)")
_, b = R.call("POST", f"{BASE}/api/v1/mp/roster/match", out_tok,
{"taskId": task_id, "realName": OUT_NAME, "idCardNo": OUT_ID})
check("门禁不通过(提示信息统一、不透露名册里有什么)", b.get("code") != 0,
json.dumps(b, ensure_ascii=False)[:200])
check("失败提示里**没有**列出名册中的任何姓名",
IN_ROSTER_NAME not in json.dumps(b, ensure_ascii=False),
json.dumps(b, ensure_ascii=False)[:200])
# ── 2. 提交申请 ──
print("\n[2] 提交加入申请")
b = body_of("POST", "/api/v1/mp/roster/join-request", out_tok, {
"taskId": task_id, "realName": OUT_NAME, "idCardNo": OUT_ID,
"phone": OUT_PHONE, "orgHint": org["orgName"], "reason": "我是转班过来的"
})
if b.get("code") != 0:
# 学生账号可能没有 open_id(token 直连不会自动建号)→ 明确告知原因并跳过
check("提交申请(需要该学生账号有 open_id)", False,
f"code={b.get('code')} msg={b.get('msg')}")
cleanup()
return 1
req_id = b["data"]["requestId"]
check("返回 requestId", isinstance(req_id, int), f"{req_id}")
check("状态为待审(0)", b["data"]["status"] == 0, f"{b['data']}")
check("提示语说清是「等待学校审核」", "审核" in (b["data"].get("statusText") or ""),
f"{b['data'].get('statusText')}")
check("告知还剩几次机会(remainAttempts=1)", b["data"].get("remainAttempts") == 1,
f"{b['data'].get('remainAttempts')}")
check("库里证件号是 HMAC(不是明文)",
sql(f"select length(id_card_hash) from collect_join_request where id={req_id};").strip()
== "64")
check("库里**没有**证件号明文列(只有 hash + mask)",
sql(f"select id_card_mask from collect_join_request where id={req_id};").strip()
.replace("*", "").isdigit()
or "*" in sql(f"select id_card_mask from collect_join_request where id={req_id};"))
check("手机号也只存 hash + 脱敏值",
sql(f"select phone_mask from collect_join_request where id={req_id};").strip()
== "139****9009",
sql(f"select phone_mask from collect_join_request where id={req_id};"))
check("记录了申请人 uid 与 openid(通过时要用来绑定)",
sql(f"select applicant_uid from collect_join_request where id={req_id};").strip()
== "993301")
# ── 3. 申请期间:**仍然不能填报**(这条最容易被做错)──
print("\n[3] 申请期间不能填报(不绕过门禁)")
b = body_of("POST", "/api/v1/mp/submissions", out_tok, {"taskId": task_id})
check("提交申请**不等于**能建填报单", b.get("code") != 0,
f"code={b.get('code')} msg={b.get('msg')}")
# ── 4. 重复申请与已在名册 ──
print("\n[4] 重复申请 / 已在名册")
b = body_of("POST", "/api/v1/mp/roster/join-request", out_tok, {
"taskId": task_id, "realName": OUT_NAME, "idCardNo": OUT_ID, "phone": OUT_PHONE
})
check("同证件号已有待审申请 → 40038", b.get("code") == 40038,
f"code={b.get('code')} msg={b.get('msg')}")
in_tok = ensure_student(993302)
b = body_of("POST", "/api/v1/mp/roster/join-request", in_tok, {
"taskId": task_id, "realName": IN_ROSTER_NAME, "idCardNo": IN_ROSTER_ID,
"phone": "13900139008"
})
check("证件号已在名册 → 40039(引导他回去重新核验)", b.get("code") == 40039,
f"code={b.get('code')} msg={b.get('msg')}")
b = body_of("POST", "/api/v1/mp/roster/join-request", out_tok, {
"taskId": task_id, "realName": OUT_NAME, "idCardNo": OUT_ID, "phone": "139"
})
check("手机号不合法 → 40001(老师要能联系到他)", b.get("code") == 40001,
f"code={b.get('code')} msg={b.get('msg')}")
# ── 5. 管理端待办 ──
print("\n[5] 管理端待办列表")
b = body_of("GET", f"/api/v1/roster/join-requests?taskId={task_id}&status=0", at)
check("待办列表可用", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
d = b.get("data") or {}
row = next((x for x in (d.get("list") or []) if x.get("requestId") == req_id), None)
check("列表里有这条申请", row is not None, json.dumps(d, ensure_ascii=False)[:300])
if row:
check("列表里证件号是脱敏的(不含 18 位明文)",
"*" in (row.get("idCardMask") or "") and len(row.get("idCardMask") or "") <= 18,
f"{row.get('idCardMask')}")
check("列表里手机号是脱敏的", "*" in (row.get("phoneMask") or ""),
f"{row.get('phoneMask')}")
check("列表带等待时长", row.get("waitHours") is not None, f"{row.get('waitHours')}")
check("列表带自述班级与申请说明",
row.get("orgHint") == org["orgName"] and row.get("reason") == "我是转班过来的",
json.dumps(row, ensure_ascii=False)[:200])
check("counts 里有三个计数", (d.get("counts") or {}).get("pending", 0) >= 1,
json.dumps(d.get("counts"), ensure_ascii=False))
# 权限:学生不能看管理端列表
b = body_of("GET", f"/api/v1/roster/join-requests?taskId={task_id}", out_tok)
check("学生调管理端接口 → 40301 或 403",
b.get("code") in (40301, 403) or b.get("_http") == 403,
f"code={b.get('code')} http={b.get('_http')}")
# ── 6. 审批通过 ──
print("\n[6] 审批通过(建名册 + 自动绑定)")
b = body_of("POST", f"/api/v1/roster/join-requests/{req_id}/approve", at,
{"orgId": org["id"]})
check("通过成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:300])
av = b.get("data") or {}
roster_id = av.get("rosterId")
check("返回新建的名册项 id", isinstance(roster_id, int), f"{roster_id}")
check("rosterCreated=true", av.get("rosterCreated") is True, f"{av}")
check("canFillNow=true(已绑定,学生重进就能填)", av.get("canFillNow") is True, f"{av}")
check("notifySent=false(通知渠道未实现,不谎报)", av.get("notifySent") is False, f"{av}")
check("名册里真的多了一行,且已绑定到申请人",
sql(f"select bind_user_id from collect_roster where id={roster_id};").strip() == "993301",
sql(f"select bind_user_id, bind_type from collect_roster where id={roster_id};"))
check("bind_type=3(来自申请)",
sql(f"select bind_type from collect_roster where id={roster_id};").strip() == "3")
check("名册行的班级是管理员指定的那个",
sql(f"select org_id from collect_roster where id={roster_id};").strip() == str(org["id"]))
check("名册行留了来源痕迹(ext_json.source=JOIN_REQUEST)",
"JOIN_REQUEST" in sql(f"select ext_json from collect_roster where id={roster_id};"))
check("申请单回写了 roster_id 与审批人",
sql(f"select roster_id from collect_join_request where id={req_id};").strip()
== str(roster_id))
# ⭐ 通过之后:门禁必须放行(自动绑定的效果)
_, b = R.call("POST", f"{BASE}/api/v1/mp/roster/match", out_tok,
{"taskId": task_id, "realName": OUT_NAME, "idCardNo": OUT_ID})
check("⭐ 通过后门禁放行(自动绑定生效)", b.get("code") == 0,
json.dumps(b, ensure_ascii=False)[:200])
b = body_of("POST", "/api/v1/mp/submissions", out_tok, {"taskId": task_id})
check("⭐ 通过后能建填报单了(申请期间不行)", b.get("code") == 0,
json.dumps(b, ensure_ascii=False)[:200])
# ── 7. 幂等与状态边界 ──
print("\n[7] 幂等与状态边界")
b = body_of("POST", f"/api/v1/roster/join-requests/{req_id}/approve", at, {"orgId": org["id"]})
check("重复通过 → 40040(不是再建一条名册)", b.get("code") == 40040,
f"code={b.get('code')} msg={b.get('msg')}")
# ⚠️ 不能写死总行数:夹具本身导入了 3 个人(曾伊豪/赵六/钱七),
# 加上审批新建的 1 个 = 4。写死"==2"会让断言因为**夹具**而红。
# 这里验的是真正要守的两件事:申请人只有一行、重复通过不再新建行。
check("申请人在名册里**只有一行**(重复通过没再建一条)",
sql(f"select count(*) from collect_roster where task_id={task_id} "
f"and bind_user_id=993301 and deleted=0;").strip() == "1",
sql(f"select count(*) from collect_roster where task_id={task_id} "
f"and bind_user_id=993301 and deleted=0;"))
check("名册总行数 = 夹具 3 人 + 审批新建 1 人",
sql(f"select count(*) from collect_roster where task_id={task_id} and deleted=0;").strip()
== "4",
sql(f"select real_name from collect_roster where task_id={task_id} and deleted=0;")
.replace("\n", " | "))
b = body_of("POST", f"/api/v1/roster/join-requests/{req_id}/reject", at, {"reason": "信息不符"})
check("已通过再驳回 → 40042(要撤销请删名册项)", b.get("code") == 40042,
f"code={b.get('code')} msg={b.get('msg')}")
# ── 8. 驳回路径 ──
print("\n[8] 驳回路径(原因必填)")
other_tok = ensure_student(993303)
other_name, other_id = "再来同学", R.with_checksum("36102920090202006")
b = body_of("POST", "/api/v1/mp/roster/join-request", other_tok, {
"taskId": task_id, "realName": other_name, "idCardNo": other_id, "phone": "13900139007"
})
check("第二个学生也能申请", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
req2 = (b.get("data") or {}).get("requestId")
b = body_of("POST", f"/api/v1/roster/join-requests/{req2}/reject", at, {})
check("驳回不填原因 → 40001(原因要给学生看)", b.get("code") == 40001,
f"code={b.get('code')} msg={b.get('msg')}")
b = body_of("POST", f"/api/v1/roster/join-requests/{req2}/reject", at,
{"reason": "你不是本校学生", "reasonCode": "NOT_THIS_SCHOOL"})
check("驳回成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
check("驳回后状态=2", sql(f"select status from collect_join_request where id={req2};").strip()
== "2")
check("驳回原因落库(学生要能看到)",
"NOT_THIS_SCHOOL" in sql(f"select reject_reason from collect_join_request where id={req2};"))
b = body_of("GET", f"/api/v1/mp/roster/join-request?taskId={task_id}", other_tok)
mine = (b.get("data") or {}).get("list") or []
check("学生能看到自己的申请与驳回原因",
any(x.get("requestId") == req2 and x.get("rejectReason") for x in mine),
json.dumps(mine, ensure_ascii=False)[:300])
# ── 9. 次数上限 ──
print("\n[9] 申请次数上限(每人每任务 ≤2)")
b = body_of("POST", "/api/v1/mp/roster/join-request", other_tok, {
"taskId": task_id, "realName": other_name, "idCardNo": other_id, "phone": "13900139007"
})
check("被驳回后可以再申请一次(remainAttempts=0)",
b.get("code") == 0 and (b.get("data") or {}).get("remainAttempts") == 0,
json.dumps(b, ensure_ascii=False)[:200])
b = body_of("POST", "/api/v1/mp/roster/join-request", other_tok, {
"taskId": task_id, "realName": other_name, "idCardNo": other_id, "phone": "13900139007"
})
check("第 3 次 → 40037(上限就是上限)", b.get("code") == 40037,
f"code={b.get('code')} msg={b.get('msg')}")
# ── 10. 任务状态边界 ──
print("\n[10] 任务不在收集期时不受理")
_, b = R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/close", at, {})
check("任务已截止", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:150])
tok4 = ensure_student(993304)
b = body_of("POST", "/api/v1/mp/roster/join-request", tok4, {
"taskId": task_id, "realName": "截止后同学",
"idCardNo": R.with_checksum("36102920090303007"), "phone": "13900139006"
})
check("已截止 → 不受理(40008)", b.get("code") == 40008,
f"code={b.get('code')} msg={b.get('msg')}")
cleanup()
print("\n" + "=" * 74)
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
print("=" * 74)
for n, det in FAILED:
print(f" ✗ {n}\n {det}")
return 1 if FAILED else 0
def cleanup() -> None:
for tid in CREATED_TASK_IDS:
R.sql(f"delete from collect_join_request where task_id={tid};")
R.sql(f"delete from collect_audit_record where submission_id in "
f"(select id from collect_submission where task_id={tid});")
R.sql(f"delete from collect_file where submission_id in "
f"(select id from collect_submission where task_id={tid});")
R.sql(f"delete from collect_submission where task_id={tid};")
R.sql(f"delete from collect_roster where task_id={tid};")
R.sql(f"delete from collect_task where id={tid};")
R.sql("delete from collect_join_request where applicant_uid in (993301, 993302, 993303, 993304);")
R.sql("delete from collect_roster where bind_user_id in (993301, 993302, 993303, 993304);")
if os.path.exists(TMP_ROSTER):
os.remove(TMP_ROSTER)
print(" 已清理:加入申请 / 名册 / 任务 / 临时 Excel")
if __name__ == "__main__":
raise SystemExit(main())