feat(audit): 审核闭环后端(M5-01~05)—— 「谁能审谁」比「怎么审」重要

一、落地
   module/audit:entity(CollectAuditRecord) + 4 个 mapper + AuditService
                 + RejectReasonCatalog(附录 C 的 10 条,写在代码里,理由同权限码)
   两端接口:PC /api/v1/audit/*(§11)、小程序 /api/v1/mp/teacher/audit/*(§10.2~10.5,只做初审)
   待办的 MP 与 PC **共用一份 VO**(契约里那两份重合九成):做两份迟早分叉,
   而"同一个学生在手机上是红、在电脑上是绿"最伤信任。

二、★ 三件必须守住的事
   ① 权限 ≠ 范围:全校几十个班主任都有 audit:first,但只有这个班的那位能审这个学生。
      初审查询强制带 org 范围,且**空列表 = 看不见任何单子**(SQL 里显式 1 = 0,
      不是"不过滤")—— 那是这类查询最经典也最严重的写法错误。
      E2E 专门造了个"同样是班主任但没被指定"的 T2:看不到待办、打不开详情、点不动通过。
   ② 有红问题不许通过(40032),且校验是**现场重算**的:提交时落库的 red_count 是快照,
      而 OCR/识别是异步的、名册与材料规则也可能被改。"放过一份有问题材料"不可撤销。
   ③ 批量通过只带绿色项,红黄一律排除(onlyClean 服务端强制 true,传 false 不报错也不生效),
      且跳过的每一条都要说清原因 —— 否则用户只会反复点那个大按钮。

三、红黄计数:列表用快照,动作/详情现场重算并**回写快照**(自己会收敛,最多错一次)
   ⚠️ 因此 detail() 不能标 readOnly=true —— 它要写回写。标了 PG 直接拒绝 UPDATE,
   而这个 bug 只在"现场值与快照不一致"时炸,也就是最需要它工作的那一次。已踩并修。

四、⚠️ 三处文档与代码不一致,需要文档侧对齐(详见交接摘要 §一·二十七)
   1. 撤回的动作码:DDL 注释 3=批量通过,而接口设计 §11.4 写"action = 3(撤回)"。
      1~6 沿用 DDL,**撤回另取 7** —— 时间轴最不能混淆:把撤回显示成批量通过,
      会让人以为这份单子被通过了。
   2. 撤回的适用状态:§11.4 只列了状态 3;额外允许**状态 2**("班主任点错了通过"正是这个状态)。
      状态机自洽:撤初审 → 1,撤终审 → 2。**需客户确认**。
   3. 文件 source 的语义:DDL(2 班主任代传/3 管理员代传/4 整理器导入) 与 §10.6(source=4 代传)
      冲突,以 DDL 为准;代传接口(M5-06)未做,时间轴已按 source∈{2,3} 预留。

五、顺带补掉两个"没人写过"的窟窿
   · 名册行的 submit_status **全仓库只有读没有写** → 名册列表状态列恒为"未提交",
     而 REPLACE 护栏只靠 bind_user_id 在生效。现在学生侧与审核侧都写它。
     它**不是单调的**:被驳回(4)后重交必须回到 2,否则班主任一直以为他还欠一份。
   · collect_submission 少映射了 class_audit_time / admin_audit_time(DDL 里一直有)。
     口径是"**本轮**审核的状态"(驳回时清空,历史在 collect_audit_record 里一条不少)。

六、审核端看材料走单独通道
   MaterialFileService 抽出 urlOf/contentOf(调用方授权),学生端仍 /mp/files/*、
   审核端 /audit/files/*。不要为了"让管理员也能看"在学生那个方法里加 or 条件——那一刻学生端就漏了。

七、一个把「取值口径」讲清楚了的坑
   SubmissionValidator 原本走 FillService.getForm(taskId)(学生视角:先过门禁、再拿自己的草稿),
   审核人必然撞 40007 且拿到的是自己那份草稿。已抽 assembleFor(task, roster, sub),两条路径共用同一段装配。
   紧接着暴露第二层:**改提交单的列不一定改变"组装出来的值"** —— 学生没动过的字段值会**回退到名册**。
   想造红问题要用**名册里没有的字段**(E2E 最后用"本职业工作年限填 999"→ VAL-17)。

八、E2E:scripts/e2e-m5-audit.py → 87/87
   复用了既有的两个脚本工具(import e2e-m3-submit 拿学生侧造数、import e2e-m2-roster 拿服务端工具),
   不复制第二份 mint/upload。覆盖:范围隔离、详情与策略、40032 现场重算、幂等回放、
   终审、批量通过的三种跳过、驳回与升级、学生重交、撤回的 40033/40034/40301 与"不能撤两次"。
   夹具教训:身份证材料的目录默认长边下限是 1400,小图必判「清晰度偏低」=黄色警告,
   于是"绿色单"前提不成立、批量断言集体失败 —— 失败的样子像功能坏了,其实只是夹具不满足前提。

九、没做的(都在计划内):代传(M5-06)、统计(M5-09)、通知(M5-12)、批量驳回/转交(P1)。
   因为没有通知渠道,驳回响应**如实**回 notifySent=false + needManualNotice=true ——
   不能为了让契约好看而写 true,那会让"学生到底收到没有"变成没人能回答的问题。

十、回归:全量 **753 条断言、0 失败**(新增 87);verify-docs 19 项通过。
    下一步:M5 前端(班主任卡片流只右滑通过 / 学生详情 / 驳回页 / 一键通过确认 / PC 终审)。
This commit is contained in:
2026-09-29 22:29:00 +08:00
parent e6cd1049d4
commit 5213bc214f
21 changed files with 3695 additions and 22 deletions
+639
View File
@@ -0,0 +1,639 @@
# -*- coding: utf-8 -*-
"""
审核闭环 E2E(M5-01 ~ M5-05):待办列表 / 详情 / 通过 / 驳回 / 批量通过 / 撤回。
**这个脚本要守住的三件事**(按重要性):
① **谁都不能审别人班的学生**。全校几十个班主任都有 `audit:first`,
但只有这个班的班主任能审这个学生 —— "权限"与"范围"是两回事,
只查权限等于把全校学生的身份证照片交给每一个班主任。
所以这里专门造一个**没被指定**的班主任 T2:他既看不到待办、也点不动通过。
② **有红问题不许通过**(40032),而且必须**现场重算**、不是拿提交时的快照。
做法:提交成功后用 SQL 把「在读学校」清空,造出一个 VAL-16 红问题
——如果服务端读的是快照(red_count=0),这次通过就会成功,而这正是最危险的 bug。
③ **批量通过只带绿色项**:红黄一律排除,且**黄的单也要被跳过并说明原因**
(原型上"一键通过"是个大按钮,最容易被手快全选)。
跑法:python scripts/e2e-m5-audit.py
前提:后端 8080 在跑、xt-pg 在跑。脚本自己造任务/老师/学生,跑完自己清理。
"""
from __future__ import annotations
import argparse
import importlib.util
import json
import os
import subprocess
import sys
import uuid
sys.stdout.reconfigure(encoding="utf-8")
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PG_CONTAINER = os.environ.get("XT_PG_CONTAINER", "xt-pg")
PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []
# 两个"学生"(MP 端只认 token 里的 sub,不要求 sys_user 里真有这个人 —— 与 e2e-m3-* 同口径)
UID_A = 992201
UID_B = 992202
def _load(module_name: str, filename: str):
"""
直接 import 既有的 E2E 脚本,复用它们的工具与夹具。
自己再写一份 `mint`/`upload`/`form_values` 就等着两边分叉:
一边改了签名一边没改,表现是"这个脚本单独跑得过、一起跑就红"。
"""
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
S = _load("e2e_m3_submit", "e2e-m3-submit.py") # 学生侧:造草稿 / 传材料 / 提交
R = _load("e2e_m2_roster", "e2e-m2-roster.py") # 服务端工具:登录 / 导名册 / 校验位
CREATED_TASK_IDS: list[int] = []
CREATED_USER_IDS: list[int] = []
TMP_FILES: list[str] = []
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:
r = subprocess.run(
["docker", "exec", "-e", "PGPASSWORD=fill_dev_pwd", PG_CONTAINER,
"psql", "-U", "fill", "-d", "filldb", "-t", "-A", "-c", stmt],
capture_output=True, text=True, timeout=40)
return (r.stdout or r.stderr).strip()
def call(base: str, method: str, path: str, token: str | None = None,
body: dict | None = None) -> dict:
"""PC 侧调用:R.call 返回 (status, body),这里只要 body。"""
_, b = R.call(method, f"{base}{path}", token, body)
return b
def build_roster(path: str, rows: list[tuple[str, str]]) -> None:
"""两行名册(姓名 + 身份证号),报考三项按任务声明填好。"""
from openpyxl import Workbook
wb = Workbook()
ws = wb.active
ws.title = "24中技数控1班"
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"138001388{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(path)
def big_jpeg() -> bytes:
"""
一张**长边够长**的测试图。
⚠️ 不能用 e2e-m3-submit 里那张 900×1200 的小图:身份证材料项的目录默认
长边下限是 **1400**(`MaterialCatalog` 的档位默认,接口传 minLongSide 也压不过它),
低于它会被判「清晰度偏低」——那是**黄色警告**,
于是"干净单"这个前提不成立,批量通过的断言会集体失败,
而失败的样子像功能坏了。夹具不满足业务前提,是这类误判最常见的来源。
"""
from PIL import Image
import io
buf = io.BytesIO()
Image.new("RGB", (1600, 1400), (200, 210, 220)).save(buf, "JPEG", quality=90)
return buf.getvalue()
def prepare_student(tok: str, sid: int, task_id: int, fill: dict, dup_first_file: bool) -> dict:
"""
把一个学生推到"已提交"。
- 填表(只填可改字段;锁定字段的值来自名册)
- 按**表单里当前必传且条件成立**的材料逐项上传(不写死材料码)
- dup_first_file:第一项材料再传一份同一张图 → 制造一条可预期的黄色警告(VAL-09)
"""
S.call("PUT", f"/api/v1/mp/submissions/{sid}", tok, {"formData": fill})
img = big_jpeg()
values, mats = S.form_values(tok, task_id)
need = [m for m in mats if m.get("required") and S.visible(m.get("visibleWhen"), values)]
for i, m in enumerate(need):
code = m.get("materialCode")
up = S.upload(tok, sid, code, f"{m.get('materialName') or code}.jpg", img)
if up.get("code") != 0:
return {"_upload_error": up}
S.call("POST", f"/api/v1/mp/files/{up['data']['id']}/process", tok)
if dup_first_file and i == 0:
up2 = S.upload(tok, sid, code, f"{m.get('materialName') or code}-重复.jpg", img)
if up2.get("code") == 0:
S.call("POST", f"/api/v1/mp/files/{up2['data']['id']}/process", tok)
return S.sub(tok, sid, True)
def main() -> int:
ap = argparse.ArgumentParser(description="审核闭环 E2E")
ap.add_argument("--base", default=os.environ.get("XT_BASE", "http://localhost:8080"))
ap.add_argument("--tenant", default=os.environ.get("XT_TENANT", "demo"))
ap.add_argument("--password", default="Admin@123")
args = ap.parse_args()
base = args.base.rstrip("/")
suffix = uuid.uuid4().hex[:6]
print("=" * 76)
print(f"审核闭环 E2E(M5-01~05) base={base}")
print("=" * 76)
# ── 0. 夹具 ────────────────────────────────────────
print("\n[0] 准备:任务 + 名册 + 两个班主任(一个被指定、一个没被指定)")
_, b = R.login(base, "admin", args.password, args.tenant)
if b.get("code") != 0:
fail("管理员登录", f"{b}")
return 1
admin = b["data"]["token"]
ok("管理员登录")
_, tree = R.call("GET", f"{base}/api/v1/orgs/tree", admin)
classes = []
def walk(nodes):
for n in nodes or []:
if n.get("orgType") == 3:
classes.append(n)
walk(n.get("children"))
walk((tree.get("data") or []))
if not classes:
fail("取一个班级", "demo 租户里没有 orgType=3 的组织")
return 1
org_id, org_name = classes[0]["id"], classes[0]["orgName"]
ok(f"取到班级:{org_name}(id={org_id})")
teachers = {}
for i, tag in enumerate(("t1", "t2")):
uname = f"e2e{tag}{suffix}"
body = call(base, "POST", "/api/v1/users", admin, {
"username": uname, "realName": f"E2E{suffix}{tag}",
# 手机号要过唯一约束:用 suffix 算出来,别用 hash()(它是进程随机的)
"phone": "13" + str(int(suffix, 16) % 10 ** 8 + i).zfill(9),
"roleCodes": ["HEAD_TEACHER"], "userType": 3,
})
if body.get("code") != 0:
fail(f"建班主任 {tag}", f"{body}")
return 1
uid = body["data"]["id"]
pwd = body["data"]["initialPassword"]
CREATED_USER_IDS.append(uid)
_, lb = R.login(base, uname, pwd, args.tenant)
if lb.get("code") != 0:
fail(f"班主任 {tag} 登录", f"{lb}")
return 1
teachers[tag] = {"id": uid, "token": lb["data"]["token"], "name": f"E2E{suffix}{tag}"}
ok(f"两个班主任账号已建(T1 id={teachers['t1']['id']} 被指定为初审人,T2 id={teachers['t2']['id']} 不指定)")
body = call(base, "POST", "/api/v1/tasks", admin, {
"taskName": f"E2E审核 {suffix}", "collectMode": 1, "batchNo": "2007",
"majorName": "数控加工", "jobName": "车工", "certType": 1,
"startTime": "2026-09-28", "endTime": "2026-12-31",
})
if body.get("code") != 0:
fail("建任务", f"{body}")
return 1
task_id = body["data"]["id"]
CREATED_TASK_IDS.append(task_id)
fix = os.path.join(ROOT, "scripts", f"_tmp_audit_{suffix}.xlsx")
TMP_FILES.append(fix)
rows = [("审核甲", R.with_checksum("361029" + "20080101" + "011")),
("审核乙", R.with_checksum("361029" + "20080202" + "022"))]
build_roster(fix, rows)
# ⚠️ 必须显式带 orgId:名册导入的 orgId 是**整批归到哪个班**(为空则名册行不留 org_id)。
# 留空的后果不是"少个字段",而是提交单没有班级 → 审核范围判不出来(40001/40301)。
body = R.upload_roster(base, admin, task_id, fix,
{"dryRun": "false", "orgId": str(org_id)})
check("名册导入 2 人", (body.get("code") == 0 and (body.get("data") or {}).get("created") == 2),
f"{body}")
check("名册行带上了班级(审核范围要靠它)",
sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} "
f"AND org_id={org_id} AND deleted=0") == "2",
sql(f"SELECT org_id FROM collect_roster WHERE task_id={task_id} AND deleted=0"))
body = call(base, "PUT", f"/api/v1/tasks/{task_id}/orgs", admin, {
"orgs": [{"orgId": org_id, "headUserId": teachers["t1"]["id"], "expectedCount": 2}],
})
check("覆盖班级 + 指定初审人 = T1", body.get("code") == 0, f"{body}")
body = call(base, "POST", f"/api/v1/tasks/{task_id}/materials", admin, {
"materialCode": "IDCARD", "materialName": "身份证", "required": True,
"fileTypes": ["jpg", "jpeg", "png"], "maxSizeKb": 1024, "maxCount": 2,
# ⚠️ 把下限调到 1KB / 长边 100:E2E 用的小图(17KB、900×1200)会被
# 材料目录的默认下限判成「体积偏小 / 清晰度偏低」——那是**黄色警告**,
# 于是"绿色单"这个前提不成立,批量通过的断言全部跟着错。
# 这坑很值得记:断言失败的样子像功能坏了,其实只是夹具不满足前提。
"minSizeKb": 1, "minLongSide": 100,
})
check("配必传材料(身份证;下限调到 1KB 免得小图被判「体积偏小」)",
body.get("code") == 0, f"{body}")
body = call(base, "POST", f"/api/v1/tasks/{task_id}/publish", admin, {"force": True})
check("发布任务", body.get("code") == 0, f"{body}")
ids = sql(f"SELECT id, real_name FROM collect_roster WHERE task_id={task_id} AND deleted=0 "
f"ORDER BY id")
pairs = [ln.split("|") for ln in ids.splitlines() if ln]
if len(pairs) != 2:
fail("取名册行", ids)
return 1
rid_a, rid_b = int(pairs[0][0]), int(pairs[1][0])
cfg = S.read_cfg()
tok_a = S.mint(cfg["jwt"], cfg["issuer"], UID_A, 1)
tok_b = S.mint(cfg["jwt"], cfg["issuer"], UID_B, 1)
try:
# ── 1. 两个学生提交 ────────────────────────────
print("\n[1] 学生提交:甲(干净)/ 乙(有一处黄色警告)")
sql(f"UPDATE collect_roster SET bind_user_id={UID_A}, bind_type=1 WHERE id={rid_a}")
r = S.call("POST", "/api/v1/mp/submissions", tok_a, {"taskId": task_id})
sid_a = r["data"]["id"]
# 刻意改一个老师填过的字段(在读学校)→ 审核端要能看出"这不是老师填的那份"
rep_a = prepare_student(tok_a, sid_a, task_id, {
"work_years": "3", "school": "我校(学生改过)",
}, dup_first_file=False)
check("甲:提交成功", (rep_a.get("data") or {}).get("submitted") is True, f"{rep_a}")
sql(f"UPDATE collect_roster SET bind_user_id={UID_B}, bind_type=1 WHERE id={rid_b}")
r = S.call("POST", "/api/v1/mp/submissions", tok_b, {"taskId": task_id})
sid_b = r["data"]["id"]
rep_b = prepare_student(tok_b, sid_b, task_id, {"work_years": "5"}, dup_first_file=True)
check("乙:带着黄色警告提交成功(学生确认过)",
(rep_b.get("data") or {}).get("submitted") is True, f"{rep_b}")
check("乙:确实是「有黄无红」——这才轮得到批量通过的排除逻辑",
((rep_b.get("data") or {}).get("yellowCount") or 0) > 0
and ((rep_b.get("data") or {}).get("redCount") or 0) == 0, f"{rep_b.get('data')}")
st = sql(f"SELECT status || '|' || red_count || '|' || yellow_count || '|' || "
f"submit_status FROM collect_submission s "
f"JOIN collect_roster r ON r.id = s.roster_id "
f"WHERE s.id={sid_a} AND s.deleted=0").split("|")
check("甲:状态 1(待初审)、红黄都是 0", st[0] == "1" and st[1] == "0" and st[2] == "0", f"{st}")
check("**名册行的提交状态跟着走到 2(待审)**——以前没人写过它,名册列表恒显示「未提交」",
st[3] == "2", f"submit_status={st[3]}")
# ── 2. 待办列表与范围隔离 ──────────────────────
print("\n[2] 待办列表:权限 ≠ 范围")
t1, t2 = teachers["t1"]["token"], teachers["t2"]["token"]
b1 = call(base, "GET", f"/api/v1/audit/todos?node=1&taskId={task_id}", t1)
d1 = b1.get("data") or {}
check("T1(被指定的初审人)看得到 2 份", d1.get("total") == 2, f"{b1}")
check("counts.all = 2", (d1.get("counts") or {}).get("all") == 2, f"{d1.get('counts')}")
check("counts.clean = 1(只有甲是绿色)",
(d1.get("counts") or {}).get("clean") == 1, f"{d1.get('counts')}")
check("counts.abnormal = 1(乙有黄色)",
(d1.get("counts") or {}).get("abnormal") == 1, f"{d1.get('counts')}")
rows1 = {x.get("submissionId"): x for x in (d1.get("list") or [])}
ca, cb = rows1.get(sid_a) or {}, rows1.get(sid_b) or {}
check("甲 batchApprovable=true(服务端判定,前端不许自己算)",
ca.get("batchApprovable") is True, f"{ca}")
check("乙 batchApprovable=false", cb.get("batchApprovable") is False, f"{cb}")
check("卡片带姓名/班级/材料数/缩略图",
ca.get("realName") and ca.get("orgName") and (ca.get("materialCount") or 0) >= 1
and len(ca.get("materialThumbs") or []) >= 1, f"{ca}")
check("甲带「学生改过」的字段(changedFields 非空)",
"school" in (ca.get("changedFields") or []), f"{ca.get('changedFields')}")
check("等待小时数是算出来的(0.x 小时)",
isinstance(ca.get("waitHours"), (int, float)) and ca.get("waitHours") < 1, f"{ca.get('waitHours')}")
b2 = call(base, "GET", f"/api/v1/audit/todos?node=1&taskId={task_id}", t2)
check("**T2(同样是班主任、但没被指定)→ 一份都看不到**",
(b2.get("data") or {}).get("total") == 0, f"{b2}")
b3 = call(base, "GET", f"/api/v1/audit/todos?node=2&taskId={task_id}", t1)
check("T1 没有终审权限 → 403(不是空列表)", b3.get("code") == 40301, f"{b3}")
# ── 3. 详情 ────────────────────────────────────
print("\n[3] 审核详情:材料预览 + 改动明细 + 时间轴 + 能不能通过")
dt = call(base, "GET", f"/api/v1/audit/submissions/{sid_a}", t1)
d = dt.get("data") or {}
check("T1 打得开", dt.get("code") == 0, f"{dt}")
dt2 = call(base, "GET", f"/api/v1/audit/submissions/{sid_a}", t2)
check("**T2 打不开同一份(403)**", dt2.get("code") == 40301, f"{dt2}")
check("姓名与脱敏证件号(不给完整号)",
d.get("realName") == "审核甲" and "****" in (d.get("idCardMask") or ""), f"{d.get('realName')}/{d.get('idCardMask')}")
mats = d.get("materials") or []
idcard = next((m for m in mats if m.get("materialCode") == "IDCARD"), None)
check("材料按定义分组(缺的那项也会出现,不是只看传了什么)",
idcard is not None and idcard.get("done") is True, f"{mats}")
files = (idcard or {}).get("files") or []
check("文件带**预览链接**(七牛签名 URL 或本地走审核端下载通道,都行)",
bool(files) and ("token=" in (files[0].get("previewUrl") or "")
or "/api/v1/audit/files/" in (files[0].get("previewUrl") or "")),
f"{files}")
cd = d.get("changedDetail") or []
check("改动明细说得出「老师填的 vs 学生改的」",
len(cd) == 1 and cd[0].get("field") == "school"
and cd[0].get("rosterValue") == "示范职业学校"
and cd[0].get("studentValue") == "我校(学生改过)", f"{cd}")
check("approvePolicy.canApprove = true(无红)",
(d.get("approvePolicy") or {}).get("canApprove") is True, f"{d.get('approvePolicy')}")
check("approvePolicy.canBatch = true(无红无黄)",
(d.get("approvePolicy") or {}).get("canBatch") is True, f"{d.get('approvePolicy')}")
check("revokePolicy:还没有可撤回的结论",
(d.get("revokePolicy") or {}).get("canRevoke") is False, f"{d.get('revokePolicy')}")
tl = d.get("timeline") or []
check("时间轴是**倒序**:最近的在最前,最早的一格是「学生提交」",
tl and tl[-1].get("node") == 0 and tl[-1].get("nodeText") == "学生提交", f"{tl}")
db = call(base, "GET", f"/api/v1/audit/submissions/{sid_b}", t1)
check("乙的 canBatch = false(有黄色)",
((db.get("data") or {}).get("approvePolicy") or {}).get("canBatch") is False,
f"{(db.get('data') or {}).get('approvePolicy')}")
# ── 4. 红问题不许通过(现场重算)───────────────
print("\n[4] 有红问题不许通过 —— 而且必须**现场重算**,不是提交时的快照")
sql(f"UPDATE collect_submission SET school=NULL WHERE id={sid_a}") # VAL-16:在校生缺在读学校
ra = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/approve", t1,
{"actionSource": "BUTTON"})
check("**清空必填项后通过 → 40032**(若读快照会误判为可过)",
ra.get("code") == 40032, f"{ra}")
check("提示里说了是哪一类问题", "在读学校" in (ra.get("msg") or ""), f"{ra.get('msg')}")
d4b = call(base, "GET", f"/api/v1/audit/submissions/{sid_a}", t1)
d4 = d4b.get("data") or {}
check("详情里的红计数也变成 1(现场值与快照已对齐)",
d4.get("redCount") == 1, f"{d4b}")
check("canApprove 变成 false,并给出原因",
(d4.get("approvePolicy") or {}).get("canApprove") is False
and "红色" in ((d4.get("approvePolicy") or {}).get("reason") or ""),
f"{d4.get('approvePolicy')}")
check("**详情这一读顺手把快照回写了**(列表下次读到的是现场值)",
sql(f"SELECT red_count FROM collect_submission WHERE id={sid_a}") == "1",
sql(f"SELECT red_count FROM collect_submission WHERE id={sid_a}"))
sql(f"UPDATE collect_submission SET school='示范职业学校' WHERE id={sid_a}")
# ── 5. 初审通过 ────────────────────────────────
print("\n[5] 初审通过:1 → 2,并留痕")
ra = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/approve", t1,
{"actionSource": "SWIPE", "remark": "材料齐全"})
da = ra.get("data") or {}
check("通过成功", ra.get("code") == 0, f"{ra}")
check("状态 → 2 且说明下一步是「等待管理员终审」",
da.get("status") == 2 and da.get("nextNode") == "ADMIN_AUDIT", f"{da}")
check("replayed=false(这是真审了一次)", da.get("replayed") is False, f"{da}")
sql_status = sql(f"SELECT status || '|' || (class_audit_by IS NOT NULL)::text || '|' || "
f"(class_audit_time IS NOT NULL)::text FROM collect_submission WHERE id={sid_a}")
check("库里:状态 2 + 初审人/时间都写了", sql_status == "2|true|true", f"{sql_status}")
check("审核记录写了一条(audit_node=CLASS_AUDIT, action=1)",
sql(f"SELECT audit_node || '|' || action || '|' || auditor_id FROM collect_audit_record "
f"WHERE submission_id={sid_a} ORDER BY id DESC LIMIT 1")
== f"CLASS_AUDIT|1|{teachers['t1']['id']}",
sql(f"SELECT audit_node || '|' || action FROM collect_audit_record WHERE submission_id={sid_a}"))
ra2 = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/approve", t1, {})
check("**重复点通过是幂等的**(返回 replayed=true,不报错、不重复写记录)",
ra2.get("code") == 0 and (ra2.get("data") or {}).get("replayed") is True, f"{ra2}")
check("幂等回放没有多写一条记录",
sql(f"SELECT count(*) FROM collect_audit_record WHERE submission_id={sid_a}") == "1",
sql(f"SELECT count(*) FROM collect_audit_record WHERE submission_id={sid_a}"))
# ── 6. 终审 ────────────────────────────────────
print("\n[6] 终审:2 → 3")
bt = call(base, "GET", f"/api/v1/audit/todos?node=2&taskId={task_id}", admin)
check("管理员终审列表看到 1 份", (bt.get("data") or {}).get("total") == 1, f"{bt}")
rfin = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/approve", admin,
{"finalOpinion": "同意认定", "needArchive": False})
check("终审通过 → 3", (rfin.get("data") or {}).get("status") == 3, f"{rfin}")
st = sql(f"SELECT status || '|' || (admin_audit_by IS NOT NULL)::text FROM collect_submission "
f"WHERE id={sid_a}")
check("库里状态 3 + 终审人已写", st.startswith("3|"), f"{st}")
check("名册行 → 3(已通过)",
sql(f"SELECT submit_status FROM collect_roster WHERE id={rid_a}") == "3",
sql(f"SELECT submit_status FROM collect_roster WHERE id={rid_a}"))
check("库里有 2 条审核记录(初审 + 终审);时间轴上的「学生提交」是合成条目",
sql(f"SELECT count(*) FROM collect_audit_record WHERE submission_id={sid_a}") == "2",
sql(f"SELECT count(*) FROM collect_audit_record WHERE submission_id={sid_a}"))
# ── 7. 批量通过:只带绿色项 ────────────────────
print("\n[7] 批量通过:红黄一律排除(铁律)")
# ① 乙此刻是「有黄无红」→ 必须被跳过,且要说清是黄色
rb = call(base, "POST", "/api/v1/audit/submissions/batch-approve", t1, {
"submissionIds": [sid_b], "onlyClean": False,
})
db_ = rb.get("data") or {}
check("onlyClean 传 false 也不生效(服务端强制为 true)", rb.get("code") == 0, f"{rb}")
details = {x.get("submissionId"): x for x in (db_.get("details") or [])}
check("**有黄的单被跳过**:approved=0",
db_.get("approved") == 0 and db_.get("skipped") == 1, f"{db_}")
check("跳过原因说清了「黄色提醒需单独确认」(不是一句「跳过」)",
"黄色" in (details.get(sid_b, {}).get("reason") or ""), f"{details.get(sid_b)}")
# ② 把那份重复件停用 → 乙变成绿色 → 这次应该真的通过
sql(f"UPDATE collect_file SET is_current=0 WHERE submission_id={sid_b} AND id IN "
f"(SELECT id FROM collect_file WHERE submission_id={sid_b} ORDER BY id DESC LIMIT 1)")
rb2 = call(base, "POST", "/api/v1/audit/submissions/batch-approve", t1, {
"submissionIds": [sid_b], "onlyClean": True,
})
db2 = rb2.get("data") or {}
check("清掉重复件后:approved=1(乙过初审)", db2.get("approved") == 1, f"{rb2}")
check("批量通过记的是 action=3(批量通过),不是 1",
sql(f"SELECT action FROM collect_audit_record WHERE submission_id={sid_b} "
f"ORDER BY id DESC LIMIT 1") == "3",
sql(f"SELECT action FROM collect_audit_record WHERE submission_id={sid_b}"))
check("批量通过后状态 → 2",
sql(f"SELECT status FROM collect_submission WHERE id={sid_b}") == "2",
sql(f"SELECT status FROM collect_submission WHERE id={sid_b}"))
# ③ 把乙弄成红色 → 红色单同样必须被跳过
# ⚠️ 两个坑,都值得记:
# ① 先把状态放回 1:批量通过之后乙已经在"待终审"了,那时初审人再点它,
# 跳过原因会是「你没有终审权限」——正确但**不对题**,会掩盖这条用例要验的东西;
# ② 红问题要用**名册里没有的字段**来造(这里用「本职业工作年限」填 999 → VAL-17)。
# 一开始我用「在读学校」清空列,结果**根本不会触发**——
# 因为组装表单值时"学生没动过"的字段会**回退到名册值**(见 FillService.assemble),
# 提交单里那列是不是 null 并不重要。这正是"校验收敛到同一份取值口径"的体现。
sql(f"UPDATE collect_submission SET status=1, work_years=999 WHERE id={sid_b}")
rb3 = call(base, "POST", "/api/v1/audit/submissions/batch-approve", t1,
{"submissionIds": [sid_b]})
details3 = {x.get("submissionId"): x for x in ((rb3.get("data") or {}).get("details") or [])}
check("**有红的单被跳过**:approved=0",
(rb3.get("data") or {}).get("approved") == 0, f"{rb3}")
check("跳过原因说清了「存在红色问题,需单独审核」",
"红色" in (details3.get(sid_b, {}).get("reason") or ""), f"{details3.get(sid_b)}")
sql(f"UPDATE collect_submission SET work_years=5, status=2 WHERE id={sid_b}")
# ── 8. 驳回 ────────────────────────────────────
print("\n[8] 驳回:原因必填 + 升级 + 学生重交")
# 让乙回到待初审:管理员先驳回(终审驳回直接生效,可越过初审)
sql(f"UPDATE collect_submission SET status=2 WHERE id={sid_b}")
r_no = call(base, "POST", f"/api/v1/audit/submissions/{sid_b}/reject", admin, {})
check("不填原因 → 拒绝", r_no.get("code") != 0, f"{r_no}")
r_bad = call(base, "POST", f"/api/v1/audit/submissions/{sid_b}/reject", admin,
{"reasonCode": "R404"})
check("原因码不在字典里 → 拒绝", r_bad.get("code") != 0, f"{r_bad}")
r_999 = call(base, "POST", f"/api/v1/audit/submissions/{sid_b}/reject", admin,
{"reasonCode": "R999"})
check("R999 不填说明 → 拒绝(学生收到一条「其他」等于没收到)",
r_999.get("code") != 0, f"{r_999}")
rej = call(base, "POST", f"/api/v1/audit/submissions/{sid_b}/reject", admin,
{"reasonCode": "R002", "reasonText": "名册上是 2008 年", "materialCodes": ["IDCARD"]})
dj = rej.get("data") or {}
check("驳回成功 → 状态 4", dj.get("status") == 4, f"{rej}")
check("notifySent=false 且 needManualNotice=true(一期没有通知渠道,**如实说**)",
dj.get("notifySent") is False and dj.get("needManualNotice") is True, f"{dj}")
check("第 1 次驳回,escalated=false", dj.get("escalated") is False, f"{dj}")
st = sql(f"SELECT status || '|' || reject_count || '|' || last_reject_reason || '|' || "
f"(SELECT submit_status FROM collect_roster WHERE id={rid_b}) "
f"FROM collect_submission WHERE id={sid_b}")
check("库里:状态 4 + 次数 1 + 原因 JSON(含码与文案)",
st.startswith("4|1|") and "R002" in st and "名册上是 2008 年" in st, f"{st}")
check("名册行 → 4(已驳回)", st.rstrip().endswith("|4"), f"{st}")
check("驳回记录写了一条(action=2 + 原因码 + 材料项)",
sql(f"SELECT action || '|' || reason_code || '|' || material_codes "
f"FROM collect_audit_record WHERE submission_id={sid_b} ORDER BY id DESC LIMIT 1")
== "2|R002|IDCARD",
sql(f"SELECT action || '|' || reason_code FROM collect_audit_record WHERE submission_id={sid_b}"))
# 升级:把次数改成 2,再驳一次就是第 3 次
sql(f"UPDATE collect_submission SET status=2, reject_count=2 WHERE id={sid_b}")
rej2 = call(base, "POST", f"/api/v1/audit/submissions/{sid_b}/reject", admin,
{"reasonCode": "R005"})
check("第 3 次驳回 → escalated=true(达上限要升级给管理员)",
(rej2.get("data") or {}).get("escalated") is True, f"{rej2}")
# 学生重新提交 → 回到待初审,名册状态也要回来
sql(f"UPDATE collect_submission SET reject_count=3 WHERE id={sid_b}")
r_re = S.sub(tok_b, sid_b, True)
check("学生被驳回后可以重新提交", (r_re.get("data") or {}).get("submitted") is True, f"{r_re}")
st = sql(f"SELECT status || '|' || (SELECT submit_status FROM collect_roster WHERE id={rid_b}) "
f"FROM collect_submission WHERE id={sid_b}")
check("**状态回到 1,名册行回到 2(待审)**——不回退的话班主任会一直以为他还欠一份",
st == "1|2", f"{st}")
# ── 9. 撤回 ────────────────────────────────────
print("\n[9] 撤回:只能撤最近一次通过;已导出/已归档不可撤")
d9 = call(base, "GET", f"/api/v1/audit/submissions/{sid_a}", admin).get("data") or {}
rp = d9.get("revokePolicy") or {}
check("管理员看得到可撤回的结论(最近一条 = 终审那条)",
rp.get("canRevoke") is True and rp.get("auditRecordId"), f"{rp}")
admin_rec = rp.get("auditRecordId")
t2_rev = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/revoke", t2,
{"auditRecordId": admin_rec, "reason": "越权试试"})
check("T2(无撤回权限)→ 403", t2_rev.get("code") == 40301, f"{t2_rev}")
# 导出过 → 40033
sql(f"INSERT INTO collect_export_task (tenant_id, task_id, export_name, status, "
f"create_time) VALUES (1, {task_id}, 'E2E假导出', 2, now())")
r_exp = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/revoke", admin,
{"auditRecordId": admin_rec, "reason": "导出后还想撤"})
check("**已导出打包 → 40033**(已交付的数据不可追溯修改)",
r_exp.get("code") == 40033, f"{r_exp}")
sql(f"DELETE FROM collect_export_task WHERE task_id={task_id} AND export_name='E2E假导出'")
# 归档 → 40034
sql(f"UPDATE collect_task SET status=3 WHERE id={task_id}")
r_arc = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/revoke", admin,
{"auditRecordId": admin_rec, "reason": "归档后还想撤"})
check("已归档 → 40034", r_arc.get("code") == 40034, f"{r_arc}")
sql(f"UPDATE collect_task SET status=1 WHERE id={task_id}")
r_ok = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/revoke", admin,
{"auditRecordId": admin_rec, "reason": "发错了人,撤回重审"})
check("撤回终审结论 → 回到 2(待终审)",
(r_ok.get("data") or {}).get("status") == 2, f"{r_ok}")
check("撤回也留痕:多了一条 action=7(撤回)的记录",
sql(f"SELECT count(*) FROM collect_audit_record WHERE submission_id={sid_a} "
f"AND action=7") == "1",
sql(f"SELECT action FROM collect_audit_record WHERE submission_id={sid_a} ORDER BY id"))
check("时间轴里能看出「谁撤回了哪一条」",
"撤回记录 #" in sql(f"SELECT reason_text FROM collect_audit_record WHERE "
f"submission_id={sid_a} AND action=7"),
sql(f"SELECT reason_text FROM collect_audit_record WHERE submission_id={sid_a} AND action=7"))
# 撤回后再撤回同一条 → 已经是最新的那条被撤了,不能重复撤
r_twice = call(base, "POST", f"/api/v1/audit/submissions/{sid_a}/revoke", admin,
{"auditRecordId": admin_rec, "reason": "再撤一次"})
check("同一条不能撤两次(它已经不是「最近一次结论」了)",
r_twice.get("code") != 0, f"{r_twice}")
# ── 10. 字典与边界 ─────────────────────────────
print("\n[10] 驳回原因字典与边界")
rc = call(base, "GET", "/api/v1/audit/reason-codes", t1)
items = (rc.get("data") or [])
check("10 条原因码(PRD 附录 C)", len(items) == 10, f"{len(items)}")
check("R999 标了 requiresText=true",
any(x.get("code") == "R999" and x.get("requiresText") is True for x in items), f"{items[:2]}")
check("文案都是人话(不是 R001 这种码,也不是空)",
all(x.get("text") and x.get("text") != x.get("code") for x in items), f"{items}")
r_over = call(base, "POST", "/api/v1/audit/submissions/batch-approve", t1,
{"submissionIds": list(range(1, 300))})
check("批量超过 200 条 → 40011", r_over.get("code") == 40011, f"{r_over}")
r_none = call(base, "POST", "/api/v1/audit/submissions/batch-approve", t1,
{"submissionIds": []})
check("空列表 → 明确报错(不是静默成功)", r_none.get("code") != 0, f"{r_none}")
# 学生不该能碰审核接口
r_stu = S.call("GET", "/api/v1/audit/todos?node=1", tok_a)
check("学生拿自己的 token 调审核接口 → 40301", r_stu.get("code") == 40301, f"{r_stu}")
finally:
print("\n[11] 清理")
for tid in CREATED_TASK_IDS:
sql(f"DELETE FROM collect_audit_record WHERE task_id={tid}")
sql(f"DELETE FROM collect_file WHERE task_id={tid}")
sql(f"DELETE FROM collect_submission WHERE task_id={tid}")
sql(f"DELETE FROM collect_roster WHERE task_id={tid}")
sql(f"DELETE FROM collect_material_def WHERE task_id={tid}")
sql(f"DELETE FROM collect_task_org WHERE task_id={tid}")
sql(f"DELETE FROM collect_export_task WHERE task_id={tid}")
sql(f"DELETE FROM collect_task WHERE id={tid}")
for uid in CREATED_USER_IDS:
sql(f"DELETE FROM sys_user_role WHERE user_id={uid}")
sql(f"DELETE FROM sys_user WHERE id={uid}")
for p in TMP_FILES:
try:
os.remove(p)
except OSError:
pass
left = sql(f"SELECT count(*) FROM collect_task WHERE task_name LIKE 'E2E审核%'")
ok(f"已清理(残留任务 {left} 个)")
print("\n" + "=" * 76)
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
for n, dd in FAILED:
print(f" · {n}\n {dd}")
print("=" * 76)
return 1 if FAILED else 0
if __name__ == "__main__":
raise SystemExit(main())