lovife
30c3714d35
新增：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。
451 lines
21 KiB
Python
451 lines
21 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""班主任代传材料 E2E（《03-接口设计》§10.6 + §4.6 教师端文件访问）。
|
||
|
||
真实场景：学生不会弄（照片在家长手机里 / 不会压缩）、或者人不在。
|
||
没有这条通道时，老师只能"让学生把照片发给自己，再让学生登录传上来" ——
|
||
而学生之所以要老师帮忙，往往就是因为他弄不了。
|
||
|
||
## 三道必须守住的闸门
|
||
|
||
① **范围**：只能给**自己负责的班**的学生代传。不校验的话，
|
||
任何老师能给任何学生塞材料 —— 那等于伪造认定材料；
|
||
② **状态**：只能给**草稿**状态的填报单代传。已提交的单子塞材料会让
|
||
"审到一半的材料清单"变了，而审核人看的是他点开那一刻的内容；
|
||
③ **留痕**：`source=2` + 代传人 + 原因 + 操作日志，且**学生端可见** ——
|
||
学生必须能知道"这份不是我传的，是李老师帮我传的"，
|
||
否则他在被驳回时会一头雾水（"我什么时候传过这张照片？"）。
|
||
|
||
## 还要验的一件事：`source` 以前从来没被写过
|
||
|
||
V12 之前 `collect_file.source` 全靠列默认值（库里 10 行全是 1），
|
||
所以"自己传的"与"代传的"在数据上**完全区分不出来**。
|
||
本脚本会断言代传之后 source 真的变成了 2。
|
||
|
||
跑法：python scripts/e2e-m5-proxy.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]] = []
|
||
CREATED_TASK_IDS: list[int] = []
|
||
TMP_ROSTER = os.path.join(ROOT, "scripts", "_e2e-proxy-roster.xlsx")
|
||
# 一张足够小的真 JPEG（1x1 白点）—— 上传要过格式幻数检查，不能是随便的字节
|
||
TINY_JPEG = os.path.join(ROOT, "scripts", "_e2e-proxy.jpg")
|
||
|
||
T1 = 993501 # 班主任甲（A 班初审人）
|
||
T2 = 993502 # 班主任乙（不负责 A 班）
|
||
S1 = 993511 # A 班学生
|
||
|
||
|
||
def _load(module_name: str, filename: str):
|
||
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")
|
||
S = _load("e2e_m3_submit", "e2e-m3-submit.py")
|
||
A = _load("e2e_m5_audit", "e2e-m5-audit.py")
|
||
|
||
BASE = "http://localhost:8080"
|
||
|
||
|
||
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:
|
||
_, out = R.sql(stmt)
|
||
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 mp(uid: int) -> str:
|
||
cfg = R.read_dev_config()
|
||
return R.mint_mp_token(cfg["jwt_secret"], cfg["issuer"], uid, 1, ["STUDENT"], 4)
|
||
|
||
|
||
def ensure_teacher(uid: int, username: str, name: str, org_id: int) -> str:
|
||
sql(f"""
|
||
INSERT INTO sys_user (id, tenant_id, username, real_name, password, org_id, user_type, status)
|
||
VALUES ({uid}, 1, '{username}', '{name}',
|
||
'$2a$10$6NB7sObvoHRwzhLJwZ2CHOAbdECgMJGiiTeMY6.zo3IJyd8TEKLEO', {org_id}, 3, 1)
|
||
ON CONFLICT (id) DO UPDATE SET real_name = EXCLUDED.real_name,
|
||
org_id = EXCLUDED.org_id,
|
||
username = EXCLUDED.username,
|
||
password = EXCLUDED.password,
|
||
user_type = EXCLUDED.user_type,
|
||
status = 1
|
||
""")
|
||
role_id = sql("select id from sys_role where tenant_id=1 and role_code='HEAD_TEACHER' "
|
||
"and deleted=0 limit 1;").strip()
|
||
sql(f"delete from sys_user_role where user_id={uid};")
|
||
sql(f"INSERT INTO sys_user_role (user_id, role_id, tenant_id) VALUES ({uid}, {role_id}, 1);")
|
||
_, b = R.login(BASE, username, "Admin@123", "demo")
|
||
if b.get("code") != 0:
|
||
raise SystemExit(f" 班主任 {username} 登录失败：{b}")
|
||
return b["data"]["token"]
|
||
|
||
|
||
def upload_proxy(token: str, sid: int, material: str, reason: str | None,
|
||
fname: str = "代传照片.jpg") -> dict:
|
||
"""用 curl 发 multipart（urllib 手搓 multipart 容易出错）。"""
|
||
import subprocess
|
||
args = ["curl", "-s", "-X", "POST",
|
||
f"{BASE}/api/v1/mp/teacher/proxy-upload",
|
||
"-H", f"Authorization: Bearer {token}",
|
||
"-F", f"submissionId={sid}",
|
||
"-F", f"materialCode={material}",
|
||
"-F", f"file=@{TINY_JPEG};filename={fname};type=image/jpeg"]
|
||
if reason is not None:
|
||
args += ["-F", f"proxyReason={reason}"]
|
||
r = subprocess.run(args, capture_output=True, text=True, encoding="utf-8", timeout=60)
|
||
try:
|
||
return json.loads(r.stdout)
|
||
except Exception:
|
||
return {"code": -1, "msg": (r.stdout or r.stderr or "")[:300]}
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="班主任代传 E2E")
|
||
ap.parse_args()
|
||
|
||
print("=" * 74)
|
||
print(" 班主任代传材料 E2E（§10.6 / §4.6）")
|
||
print("=" * 74)
|
||
|
||
_, adm = R.login(BASE, "admin", "Admin@123", "demo")
|
||
if adm.get("code") != 0:
|
||
print(f" ✗ 管理员登录失败：{adm}")
|
||
return 2
|
||
at = adm["data"]["token"]
|
||
_make_jpeg()
|
||
|
||
# ── 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 [])
|
||
org_a, org_b = classes[0], classes[1]
|
||
|
||
t1 = ensure_teacher(T1, f"proxyT1{suffix}", "代传班主任甲", org_a["id"])
|
||
t2 = ensure_teacher(T2, f"proxyT2{suffix}", "代传班主任乙", org_b["id"])
|
||
ok("两个班主任账号就绪")
|
||
|
||
R.call("PUT", f"{BASE}/api/v1/tasks/{task_id}/orgs", at, {"orgs": [
|
||
{"orgId": org_a["id"], "expectedCount": 1, "headUserId": T1},
|
||
{"orgId": org_b["id"], "expectedCount": 0, "headUserId": T2}
|
||
]})
|
||
R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/materials/apply-template", at,
|
||
{"templateCode": "TPL_SCHOOL_ROLL"})
|
||
|
||
rows = [("代传甲", R.with_checksum("36102920080101061"))]
|
||
_build_roster(rows)
|
||
body = R.upload_roster(BASE, at, task_id, TMP_ROSTER,
|
||
extra={"dryRun": "false", "mode": "APPEND", "orgId": org_a["id"]})
|
||
check("名册导入成功", body.get("code") == 0, json.dumps(body, ensure_ascii=False)[:200])
|
||
R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/publish", at, {"force": True})
|
||
|
||
tok = mp(S1)
|
||
_, b = R.call("POST", f"{BASE}/api/v1/mp/roster/match", tok,
|
||
{"taskId": task_id, "realName": rows[0][0], "idCardNo": rows[0][1]})
|
||
check("学生门禁绑定成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:200])
|
||
_, b = R.call("POST", f"{BASE}/api/v1/mp/submissions", tok, {"taskId": task_id})
|
||
sid = b["data"]["id"]
|
||
check("学生建了草稿", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:150])
|
||
|
||
# ── 1. 代传上下文（界面要一直显示"替谁传"）──
|
||
print("\n[1] 代传前：我到底在替谁传")
|
||
ctx = body_of("GET", f"/api/v1/mp/teacher/submissions/{sid}/proxy-context", t1).get("data") or {}
|
||
check("上下文接口可用", ctx.get("submissionId") == sid,
|
||
json.dumps(ctx, ensure_ascii=False)[:200])
|
||
check("带学生姓名（防止传给上一个人）", ctx.get("realName") == rows[0][0],
|
||
f"{ctx.get('realName')}")
|
||
check("带脱敏证件号（老师要能确认是这个人）", bool(ctx.get("idCardMask")),
|
||
f"{ctx.get('idCardMask')}")
|
||
check("标出可否代传", ctx.get("canProxy") is True, f"{ctx.get('canProxy')}")
|
||
ctx2 = body_of("GET", f"/api/v1/mp/teacher/submissions/{sid}/proxy-context", t2)
|
||
check("⭐ 别的班的老师看不到这个上下文（40301）", ctx2.get("code") == 40301,
|
||
f"code={ctx2.get('code')} msg={ctx2.get('msg')}")
|
||
|
||
# ── 2. 代传：一次 multipart 搞定 ──
|
||
print("\n[2] 代传一份材料")
|
||
# ⚠️ 代传用 IDCARD 而不是 PHOTO1：PHOTO1 只有 1 份上限，
|
||
# 如果代传占了它，后面"学生补齐并提交"就会撞 40001（只能传 1 份）——
|
||
# 那不是被测代码的问题，是夹具自己占掉了名额。
|
||
up = upload_proxy(t1, sid, "IDCARD", "学生说照片在家长手机里，我帮他传")
|
||
check("代传成功", up.get("code") == 0, json.dumps(up, ensure_ascii=False)[:300])
|
||
fv = up.get("data") or {}
|
||
file_id = fv.get("fileId")
|
||
check("返回 fileId", isinstance(file_id, int), f"{file_id}")
|
||
check("返回代传人 id", fv.get("proxyBy") == T1, f"{fv.get('proxyBy')}")
|
||
check("⭐ 返回代传人姓名（审核时间轴要显示「由李老师代传」）",
|
||
"代传班主任甲" in (fv.get("proxyByName") or ""), f"{fv.get('proxyByName')}")
|
||
check("返回可显示的来源文案", "代传" in (fv.get("sourceText") or ""),
|
||
f"{fv.get('sourceText')}")
|
||
|
||
# ── 3. 留痕：source / proxy_by / proxy_reason / 操作日志 ──
|
||
print("\n[3] 留痕（契约 §10.6：代传必须留痕）")
|
||
check("⭐ source = 2（班主任代传）—— V12 之前这一列从来没被写过",
|
||
sql(f"select source from collect_file where id={file_id};").strip() == "2",
|
||
sql(f"select source from collect_file where id={file_id};"))
|
||
check("⭐ 记了代传人", sql(f"select proxy_by from collect_file where id={file_id};").strip()
|
||
== str(T1), sql(f"select proxy_by from collect_file where id={file_id};"))
|
||
check("⭐ 记了代传原因",
|
||
"家长手机" in sql(f"select coalesce(proxy_reason,'') from collect_file where id={file_id};"),
|
||
sql(f"select proxy_reason from collect_file where id={file_id};"))
|
||
check("⭐ 操作日志里有 PROXY_UPLOAD",
|
||
sql("select count(*) from sys_operation_log where action='PROXY_UPLOAD' "
|
||
f"and target_id={file_id};").strip() == "1",
|
||
sql("select action, target_type, extra_json from sys_operation_log "
|
||
"where action='PROXY_UPLOAD' order by id desc limit 1;"))
|
||
check("日志的 target 指向文件（可按对象查）",
|
||
sql(f"select target_type from sys_operation_log where action='PROXY_UPLOAD' "
|
||
f"and target_id={file_id} order by id desc limit 1;").strip() == "FILE")
|
||
check("文件名也按规范件模板改过（代传不能绕过规范化）",
|
||
sql(f"select standard_name from collect_file where id={file_id};").strip() != "",
|
||
sql(f"select original_name, standard_name from collect_file where id={file_id};"))
|
||
|
||
# ── 4. 学生端可见 ──
|
||
print("\n[4] 学生端可见（不然他被驳回时会一头雾水）")
|
||
mine = body_of("GET", f"/api/v1/mp/submissions/{sid}/progress", tok).get("data") or {}
|
||
mats = mine.get("materials") or []
|
||
photo = next((m for m in mats if m.get("materialCode") == "IDCARD"), None)
|
||
check("学生能在进度页看到这份材料", (photo or {}).get("done") is True,
|
||
json.dumps(photo, ensure_ascii=False))
|
||
check("进度条不计入「缺材料」（代传的也算已交）",
|
||
all(m.get("issueLevel", 0) < 3 for m in mats if m.get("done")),
|
||
json.dumps([(m.get("materialCode"), m.get("issueLevel")) for m in mats],
|
||
ensure_ascii=False))
|
||
|
||
# ── 5. 闸门二：已提交不能代传 ──
|
||
print("\n[5] 闸门：已提交的填报单不能代传")
|
||
res = A.prepare_student(tok, sid, task_id, {
|
||
"work_unit": "无", "is_retake": False, "is_student": True,
|
||
"school": "示范职业学校", "major": "数控加工", "enroll_date": "2024-09-01"
|
||
}, dup_first_file=False)
|
||
check("学生补齐并提交", (res.get("data") or {}).get("submitted") is True,
|
||
json.dumps(res, ensure_ascii=False)[:250])
|
||
up2 = upload_proxy(t1, sid, "IDCARD", None)
|
||
check("⭐ 已提交后代传被拒（40030 已进入审核）", up2.get("code") == 40030,
|
||
f"code={up2.get('code')} msg={up2.get('msg')}")
|
||
check("提示说清要先撤回",
|
||
"撤回" in (up2.get("msg") or ""), f"{up2.get('msg')}")
|
||
|
||
# ── 6. 闸门一：范围 ──
|
||
print("\n[6] 闸门：只能给自己负责的班代传")
|
||
# 用乙的身份代传甲班学生的材料
|
||
up3 = upload_proxy(t2, sid, "IDCARD", None)
|
||
check("⭐ 别的班的老师代传 → 40301", up3.get("code") == 40301,
|
||
f"code={up3.get('code')} msg={up3.get('msg')}")
|
||
up4 = upload_proxy(t1, 999999999, "IDCARD", None)
|
||
check("不存在的填报单 → 明确报错", up4.get("code") != 0,
|
||
f"code={up4.get('code')} msg={up4.get('msg')}")
|
||
up5 = upload_proxy(t1, sid, "NOT_EXIST_MATERIAL", None)
|
||
check("任务里没有的材料码 → 明确报错", up5.get("code") != 0,
|
||
f"code={up5.get('code')} msg={up5.get('msg')}")
|
||
|
||
# ── 7. 教师端文件访问（§4.6）──
|
||
print("\n[7] 教师端看学生材料（按班级范围授权）")
|
||
pv = os.path.join(ROOT, "scripts", "_e2e-proxy-dl.jpg")
|
||
code = os.system(f'curl -s -o "{pv}" -H "Authorization: Bearer {t1}" '
|
||
f'"{BASE}/api/v1/mp/teacher/files/{file_id}/preview"')
|
||
size = os.path.getsize(pv) if os.path.exists(pv) else 0
|
||
check("⭐ 班主任能看自己班学生的材料", size > 100, f"exit={code} size={size}")
|
||
# Content-Type 必须是图片：给 octet-stream 会让"点开身份证"变成"下载一个文件"。
|
||
# ⚠️ 不能用 R.call 读这个响应 —— 它是二进制，json 解析会 UnicodeDecodeError，
|
||
# 所以用 curl 的 -D 把响应头写到临时文件再看。
|
||
hdr = os.path.join(ROOT, "scripts", "_e2e-proxy-head.txt")
|
||
os.system(f'curl -s -o NUL -D "{hdr}" -H "Authorization: Bearer {t1}" '
|
||
f'"{BASE}/api/v1/mp/teacher/files/{file_id}/preview"')
|
||
ctype = ""
|
||
if os.path.exists(hdr):
|
||
with open(hdr, encoding="utf-8", errors="replace") as f:
|
||
for line in f:
|
||
if line.lower().startswith("content-type"):
|
||
ctype = line.strip()
|
||
os.remove(hdr)
|
||
check("⭐ 返回的是图片 Content-Type（否则「点开身份证」会变成「下载文件」）",
|
||
"image/" in ctype.lower(), f"{ctype}")
|
||
code2 = os.system(f'curl -s -o "{pv}2" -H "Authorization: Bearer {t2}" '
|
||
f'"{BASE}/api/v1/mp/teacher/files/{file_id}/preview"')
|
||
body2 = ""
|
||
if os.path.exists(pv + "2"):
|
||
with open(pv + "2", encoding="utf-8", errors="replace") as f:
|
||
body2 = f.read()[:200]
|
||
os.remove(pv + "2")
|
||
check("⭐ 别的班的老师看同一份材料 → 被拒（40301）", "40301" in body2,
|
||
f"exit={code2} body={body2[:150]}")
|
||
stok = mp(S1)
|
||
_, b = R.call("GET", f"{BASE}/api/v1/mp/teacher/files/{file_id}/preview", stok)
|
||
check("学生调教师端文件接口 → 被拒", b.get("code") in (40101, 40301, 403),
|
||
f"code={b.get('code')}")
|
||
if os.path.exists(pv):
|
||
os.remove(pv)
|
||
|
||
# ── 8. 权限 ──
|
||
print("\n[8] 权限")
|
||
# ⚠️ 必须发**真正的** multipart：这个接口声明了 consumes=multipart/form-data，
|
||
# 用 JSON body 会在进 Controller 之前就 500（媒体类型不匹配），
|
||
# 那样测的是"请求形态不对"而不是"未登录"。
|
||
import subprocess
|
||
r = subprocess.run(["curl", "-s", "-X", "POST",
|
||
f"{BASE}/api/v1/mp/teacher/proxy-upload",
|
||
"-F", f"submissionId={sid}",
|
||
"-F", "materialCode=IDCARD",
|
||
"-F", f"file=@{TINY_JPEG};filename=x.jpg;type=image/jpeg"],
|
||
capture_output=True, text=True, encoding="utf-8", timeout=60)
|
||
try:
|
||
nb = json.loads(r.stdout)
|
||
except Exception:
|
||
nb = {"code": -1, "msg": (r.stdout or "")[:200]}
|
||
check("未登录代传 → 40101", nb.get("code") == 40101, f"code={nb.get('code')} msg={nb.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 _make_jpeg() -> None:
|
||
"""造一张真 JPEG：上传要过格式幻数检查，随便的字节会被 40004 拒掉。"""
|
||
import base64
|
||
# 1x1 白点 JPEG
|
||
b64 = ("/9j/4AAQSkZJRgABAQEAYABgAAD/2wBDAAgGBgcGBQgHBwcJCQgKDBQNDAsLDBkSEw8UHRofHh0a"
|
||
"HBwgJC4nICIsIxwcKDcpLDAxNDQ0Hyc5PTgyPC4zNDL/wAALCAABAAEBAREA/8QAFAABAAAAAAAA"
|
||
"AAAAAAAAAAAACf/EABQQAQAAAAAAAAAAAAAAAAAAAAD/2gAIAQEAAD8AKp//2Q==")
|
||
with open(TINY_JPEG, "wb") as f:
|
||
f.write(base64.b64decode(b64))
|
||
|
||
|
||
def _build_roster(rows: list[tuple[str, str]]) -> None:
|
||
from openpyxl import Workbook
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "代传班"
|
||
header = ["序号", "姓名", "身份证号", "学历", "联系电话", "工作单位",
|
||
"报考职业", "报考工种", "报考等级", "是否补考",
|
||
"补考理论", "补考实操", "补考综合", "是否在校生",
|
||
"在读学校", "所学专业", "入学时间"]
|
||
for c, t in enumerate(header, start=1):
|
||
ws.cell(row=1, column=c, value=t)
|
||
for i, (name, idc) in enumerate(rows):
|
||
vals = [i + 1, name, idc, "中专", f"138001344{i}", "无",
|
||
"车工", "数控车工", "四级", "否", "", "", "", "是",
|
||
"示范职业学校", "数控加工", "2024-09-01"]
|
||
for c, v in enumerate(vals, start=1):
|
||
if v != "":
|
||
ws.cell(row=i + 2, column=c, value=v)
|
||
wb.save(TMP_ROSTER)
|
||
|
||
|
||
def cleanup() -> None:
|
||
"""清理测试数据，并**把失败暴露出来**。
|
||
|
||
⚠️ 这个函数原来是静默失败的：`R.sql(...)` 的返回值没人看，
|
||
而"已发布的任务删不掉"这件事没有被处理 —— 残留的 roster 行会让
|
||
后面的脚本（班主任看板）算出错误的人数，而排查时完全看不出原因。
|
||
所以现在：按依赖顺序删 → **逐条记结果** → 末尾校验一行都没剩。
|
||
"""
|
||
report: list[str] = []
|
||
for tid in CREATED_TASK_IDS:
|
||
# 顺序不能反：先删引用它们的子表，否则会撞外键/留下孤儿
|
||
steps = [
|
||
("操作日志", f"delete from sys_operation_log where target_id in "
|
||
f"(select id from collect_file where task_id={tid});"),
|
||
("文件", f"delete from collect_file where task_id={tid};"),
|
||
("填报单", f"delete from collect_submission where task_id={tid};"),
|
||
("名册", f"delete from collect_roster where task_id={tid};"),
|
||
("班级配置", f"delete from collect_task_org where task_id={tid};"),
|
||
("材料定义", f"delete from collect_material_def where task_id={tid};"),
|
||
("任务", f"delete from collect_task where id={tid};"),
|
||
]
|
||
for label, stmt in steps:
|
||
okk, out = R.sql(stmt)
|
||
report.append(f"{label} task={tid}: {'OK' if okk else 'FAIL ' + out[:120]}")
|
||
for label, stmt in [
|
||
("代理上传日志", "delete from sys_operation_log where request_uri like '%proxy-upload%';"),
|
||
("角色绑定", "delete from sys_user_role where user_id in (993501, 993502, 993511);"),
|
||
("测试账号", "delete from sys_user where id in (993501, 993502, 993511);"),
|
||
]:
|
||
okk, out = R.sql(stmt)
|
||
report.append(f"{label}: {'OK' if okk else 'FAIL ' + out[:120]}")
|
||
|
||
# ★ 校验：一行都不该剩（这一步才是"清理真的成功了吗"的答案）
|
||
leftovers = 0
|
||
for tid in CREATED_TASK_IDS:
|
||
for tbl in ("collect_roster", "collect_submission", "collect_file"):
|
||
_, n = R.sql(f"select count(*) from {tbl} where task_id={tid};")
|
||
if n.strip() not in ("", "0"):
|
||
leftovers += int(n.strip())
|
||
report.append(f"⭐ 残留 {tbl} task={tid}: {n.strip()} 行")
|
||
report_path = os.path.join(ROOT, "scripts", "_e2e-clean-report.txt")
|
||
with open(report_path, "w", encoding="utf-8") as f:
|
||
f.write("\n".join(report))
|
||
|
||
for p in (TMP_ROSTER, TINY_JPEG):
|
||
if os.path.exists(p):
|
||
os.remove(p)
|
||
|
||
fails = [x for x in report if "FAIL" in x or "残留" in x]
|
||
if fails:
|
||
print(f" ⚠️ 清理有 {len(fails)} 处问题（详见 scripts/_e2e-clean-report.txt）：")
|
||
for x in fails[:6]:
|
||
print(f" {x}")
|
||
else:
|
||
print(" 已清理（并校验无残留）：任务 / 名册 / 填报单 / 材料 / 操作日志 / 测试账号")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|