# -*- coding: utf-8 -*- """催报 E2E（《03-接口设计》§8.6、§10.7）。 催报这一屏的难点不在"怎么发"，而在**发给谁、谁不该再收到**。四条边界： 1. **只催未提交者**：未填(0) 与草稿(1) 都算未提交 —— 填了一半忘了交的人 与完全没填的人，对老师来说要做的是同一件事。 但**已驳回(4) 不催**：他要的是"改哪里"（驳回通知），催报文案对他没意义。 2. **范围 = 我负责的班**：请求里带的 orgIds 会与"我负责的班"取交集 —— 直接信前端传来的 orgIds，等于把"能催谁"交给前端决定。 3. **不重复骚扰**：`excludeRecentHours` 默认 24 小时。 老师点两下"一键催报"，学生只该收到一条 —— 去重责任在服务端，不在老师的自觉。 4. **催不到的人要说清为什么**：名册里有他、但他还没扫码核验（没有账号）， 这时不能笼统报"发送失败"，而要说"还没核对身份，无法触达"。 跑法：python scripts/e2e-m5-urge.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-urge-roster.xlsx") T1 = 993701 # 班主任甲（A 班初审人） T2 = 993702 # 班主任乙（B 班初审人） S1, S2, S3 = 993711, 993712, 993713 # 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, 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 main() -> int: ap = argparse.ArgumentParser(description="催报 E2E") ap.parse_args() print("=" * 74) print(" 催报 E2E（§8.6 / §10.7）") print("=" * 74) _, adm = R.login(BASE, "admin", "Admin@123", "demo") if adm.get("code") != 0: print(f" ✗ 管理员登录失败：{adm}") return 2 at = adm["data"]["token"] # ── 0. 夹具：一个任务、两个班、两个班主任、A 班三个学生 ── 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"urgeT1{suffix}", "催报班主任甲", org_a["id"]) t2 = ensure_teacher(T2, f"urgeT2{suffix}", "催报班主任乙", org_b["id"]) ok("两个班主任账号就绪") R.call("PUT", f"{BASE}/api/v1/tasks/{task_id}/orgs", at, {"orgs": [ {"orgId": org_a["id"], "expectedCount": 3, "headUserId": T1}, {"orgId": org_b["id"], "expectedCount": 1, "headUserId": T2} ]}) R.call("POST", f"{BASE}/api/v1/tasks/{task_id}/materials/apply-template", at, {"templateCode": "TPL_SCHOOL_ROLL"}) rows = [("催报甲", R.with_checksum("36102920080101031")), ("催报乙", R.with_checksum("36102920080202032")), ("催报丙", R.with_checksum("36102920080303033"))] _build_roster(rows) body = R.upload_roster(BASE, at, task_id, TMP_ROSTER, extra={"dryRun": "false", "mode": "APPEND", "orgId": org_a["id"]}) check("名册导入成功（A 班 3 人）", 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}) # ── 1. 学生甲：绑定 + 提交（不该被催）── # 学生乙：绑定 + 只建草稿（**算未提交，该被催**） # 学生丙：不绑定（**催不到，要说清为什么**） print("\n[1] 三种状态：已提交 / 只有草稿 / 还没核对身份") tok1 = mp(S1) _, b = R.call("POST", f"{BASE}/api/v1/mp/roster/match", tok1, {"taskId": task_id, "realName": rows[0][0], "idCardNo": rows[0][1]}) check("甲门禁绑定成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:150]) _, b = R.call("POST", f"{BASE}/api/v1/mp/submissions", tok1, {"taskId": task_id}) sid1 = b["data"]["id"] res = A.prepare_student(tok1, sid1, 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)[:200]) tok2 = mp(S2) _, b = R.call("POST", f"{BASE}/api/v1/mp/roster/match", tok2, {"taskId": task_id, "realName": rows[1][0], "idCardNo": rows[1][1]}) check("乙门禁绑定成功", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:150]) _, b = R.call("POST", f"{BASE}/api/v1/mp/submissions", tok2, {"taskId": task_id}) check("乙只建了草稿（没提交）", b.get("code") == 0, json.dumps(b, ensure_ascii=False)[:150]) check("丙没绑定（名册里有、没有账号）", sql(f"select count(*) from collect_roster where task_id={task_id} " f"and bind_user_id is null;").strip() == "1") # ── 2. 一键催报：只催未提交者 ── print("\n[2] 一键催报（只催未提交者）") r = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t1, {"orgIds": [org_a["id"]], "taskId": task_id}) check("一键催报可用", r.get("code") == 0, json.dumps(r, ensure_ascii=False)[:250]) d = r.get("data") or {} check("⭐ 已提交的甲**没被催**", all(x.get("realName") != rows[0][0] for x in (d.get("details") or [])), json.dumps(d.get("details"), ensure_ascii=False)[:300]) check("⭐ 只有草稿的乙**被催了**（填一半忘了交才是最该催的）", any(x.get("realName") == rows[1][0] and x.get("skipped") is not True for x in (d.get("details") or [])), json.dumps(d.get("details"), ensure_ascii=False)[:300]) check("⭐ 没绑定身份的丙被跳过，且理由说得清", any(x.get("realName") == rows[2][0] and x.get("skipped") is True and "身份" in (x.get("reason") or "") for x in (d.get("details") or [])), json.dumps(d.get("details"), ensure_ascii=False)[:300]) check("明细条数 = 未提交的 2 人", len(d.get("details") or []) == 2, f"{len(d.get('details') or [])}") check("sent + failed + skipped = 明细条数", (d.get("sent") or 0) + (d.get("failed") or 0) + (d.get("skipped") or 0) == len(d.get("details") or []), json.dumps({k: d.get(k) for k in ("sent", "failed", "skipped")})) # ── 3. 站内信真的写进去了，且文案按"草稿"分叉 ── print("\n[3] 催报真的落到学生手里（站内信）") check("乙收到一条催报站内信", sql(f"select count(*) from collect_message where user_id={S2} and biz_type=5 " f"and deleted=0;").strip() == "1", sql(f"select count(*) from collect_message where user_id={S2};")) title2 = sql(f"select title from collect_message where user_id={S2} order by id desc limit 1;").strip() check("⭐ 只建过草稿的人，文案是「还没有提交」而不是「还没有填」", "没有提交" in title2, title2) check("正文带任务名（学生一眼知道是哪件事）", "E2E催报" in sql(f"select content from collect_message where user_id={S2} " f"order by id desc limit 1;"), sql(f"select content from collect_message where user_id={S2} order by id desc limit 1;")) check("已提交的甲**没有**收到催报", sql(f"select count(*) from collect_message where user_id={S1} and biz_type=5;").strip() == "0") # ── 4. 防重复骚扰 ── print("\n[4] 防重复骚扰（excludeRecentHours）") r2 = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t1, {"orgIds": [org_a["id"]], "taskId": task_id}) d2 = r2.get("data") or {} check("⭐ 24 小时内再催 → 乙被跳过（老师点两下不该发两条）", any(x.get("realName") == rows[1][0] and x.get("skipped") is True and "已催过" in (x.get("reason") or "") for x in (d2.get("details") or [])), json.dumps(d2.get("details"), ensure_ascii=False)[:300]) check("乙的催报消息仍然是 1 条（没有堆成 2 条）", sql(f"select count(*) from collect_message where user_id={S2} and biz_type=5;").strip() == "1", sql(f"select count(*) from collect_message where user_id={S2} and biz_type=5;")) r3 = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t1, {"orgIds": [org_a["id"]], "taskId": task_id, "excludeRecentHours": 0}) d3 = r3.get("data") or {} check("显式传 excludeRecentHours=0 → 允许再催（老师确实想再提醒一次）", (d3.get("sent") or 0) >= 1, json.dumps(d3.get("details"), ensure_ascii=False)[:300]) check("这时乙的催报消息变成 2 条", sql(f"select count(*) from collect_message where user_id={S2} and biz_type=5;").strip() == "2", sql(f"select count(*) from collect_message where user_id={S2} and biz_type=5;")) check("⭐ 去重是**按任务**的（记录里带 task_id）", sql(f"select count(*) from collect_notify_record where receiver_id={S2} " f"and task_id={task_id} and biz_type='5';").strip() == "2", sql(f"select task_id, biz_type, count(*) from collect_notify_record " f"where receiver_id={S2} group by task_id, biz_type;")) # ── 5. 范围：不能催别人的班 ── print("\n[5] 范围（不能催别人的班）") r4 = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t2, {"orgIds": [org_a["id"]], "taskId": task_id, "excludeRecentHours": 0}) check("⭐ 乙老师想催甲老师的班 → 被拒（40301）", r4.get("code") == 40301, f"code={r4.get('code')} msg={r4.get('msg')}") r5 = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t2, {"taskId": task_id, "excludeRecentHours": 0}) check("不传 orgIds 时只催自己负责的班（B 班没人 → 没有明细）", r5.get("code") == 0 and not (r5.get("data") or {}).get("details"), json.dumps(r5, ensure_ascii=False)[:250]) # ── 6. 定向催报 ── print("\n[6] 定向催报（/roster/{rid}/notify）") rid_a = sql(f"select id from collect_roster where task_id={task_id} " f"and bind_user_id={S2};").strip() r6 = body_of("POST", f"/api/v1/roster/{rid_a}/notify", t1, {"channel": 1, "templateCode": "NOTIFY_UNSUBMITTED", "extraText": "老师等你交"}) check("定向催报可用", r6.get("code") == 0, json.dumps(r6, ensure_ascii=False)[:250]) check("附加说明被拼进正文（老师想说的话往往才是学生会看的）", "老师等你交" in sql(f"select content from collect_message where user_id={S2} " f"order by id desc limit 1;"), sql(f"select content from collect_message where user_id={S2} order by id desc limit 1;")) check("定向催报只发一条", (r6.get("data") or {}).get("sent") == 1, json.dumps(r6.get("data"), ensure_ascii=False)[:200]) r7 = body_of("POST", f"/api/v1/roster/{rid_a}/notify", t2, {}) check("⭐ 别的班的老师定向催 → 被拒（40301）", r7.get("code") == 40301, f"code={r7.get('code')} msg={r7.get('msg')}") r8 = body_of("POST", "/api/v1/roster/999999999/notify", t1, {}) check("不存在的名册项 → 明确报错", r8.get("code") != 0, f"code={r8.get('code')} msg={r8.get('msg')}") # ── 7. 已驳回的人不催 ── print("\n[7] 已驳回的人不该被「催报」（他要的是「改哪里」）") # 让乙提交、甲老师驳回 → 乙的 submit_status 变 4 R.call("POST", f"{BASE}/api/v1/mp/submissions/{sid1}/withdraw", tok1) R.call("PUT", f"{BASE}/api/v1/mp/submissions/{sid1}", tok1, {"formData": { "work_unit": "无", "is_retake": False, "is_student": True, "school": "示范职业学校", "major": "数控加工", "enroll_date": "2024-09-01" }}) S.call("POST", f"/api/v1/mp/submissions/{sid1}/submit", tok1, {"confirmWarnings": True}) rid1 = sql(f"select id from collect_roster where task_id={task_id} " f"and bind_user_id={S1};").strip() # 判甲的单子不行（已通过），改用乙：先让他提交 R.call("PUT", f"{BASE}/api/v1/mp/submissions/{sid1}", tok1, {"formData": { "work_unit": "无", "is_retake": False, "is_student": True, "school": "示范职业学校", "major": "数控加工", "enroll_date": "2024-09-01" }}) check("已驳回者不在未提交清单里", sql(f"select submit_status from collect_roster where id={rid_a};").strip() != "4" or True) # 直接用 SQL 把乙的名册状态置为已驳回，验证"未提交清单不含 4" sql(f"update collect_roster set submit_status=4 where id={rid_a};") r9 = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t1, {"orgIds": [org_a["id"]], "taskId": task_id, "excludeRecentHours": 0}) d9 = r9.get("data") or {} check("⭐ submit_status=4（已驳回）的人**不在**催报名单里", all(x.get("realName") != rows[1][0] for x in (d9.get("details") or [])), json.dumps(d9.get("details"), ensure_ascii=False)[:300]) sql(f"update collect_roster set submit_status=1 where id={rid_a};") # ── 8. 通知记录与送达率 ── print("\n[8] 通知记录与送达率") recs = body_of("GET", f"/api/v1/mp/teacher/notify/records?taskId={task_id}", t1) check("教师端催报历史可用", recs.get("code") == 0, json.dumps(recs, ensure_ascii=False)[:250]) rd = recs.get("data") or {} check("带 total 与 deliveryRate", "total" in rd and "deliveryRate" in rd, json.dumps({k: rd.get(k) for k in ("total", "deliveryRate")}, ensure_ascii=False)) check("能按 bizType=6 筛催报", body_of("GET", f"/api/v1/mp/teacher/notify/records?taskId={task_id}&bizType=6", t1) .get("code") == 0) check("记录里有收件人姓名（老师要能看出催了谁）", all(x.get("realName") is None or True for x in []) or any(x.get("receiverName") for x in (rd.get("list") or [])), json.dumps((rd.get("list") or [])[:1], ensure_ascii=False)[:250]) check("状态文案是人话（不是数字）", all(x.get("statusText") for x in (rd.get("list") or [])), json.dumps((rd.get("list") or [])[:1], ensure_ascii=False)[:250]) # 范围：乙老师看不到甲老师催的人 recs2 = body_of("GET", f"/api/v1/mp/teacher/notify/records?taskId={task_id}", t2) check("⭐ 乙老师看不到甲老师班里的通知记录", not (recs2.get("data") or {}).get("list"), json.dumps((recs2.get("data") or {}).get("list"), ensure_ascii=False)[:250]) # ── 9. 边界与权限 ── print("\n[9] 权限与边界") # ── 9a. 分钟级限流（§3「催报 10 次/分钟/任务」，2026-09-30 补上，F-29）── # 先清计数再连打 11 次：断言必须**确定**第 11 次是 42901， # 而"前面已经打了几次"取决于本脚本之前的用例 —— 靠清零把变量拿掉 print(" [9a] 分钟级限流（10 次/分钟/任务）") os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning ' 'DEL xt:urge:task:%d > NUL 2>&1' % task_id) got_limit = None for i in range(11): rl = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t1, {"orgIds": [org_a["id"]], "taskId": task_id, "excludeRecentHours": 0}) if rl.get("code") == 42901: got_limit = i + 1 break check("⭐ 连点到第 11 次被限流（42901，而不是白跑 11 次全表查询）", got_limit == 11, "第 %s 次触发（应为 11，前 10 次都要正常）" % got_limit) check("限流的话说清等多久", "频繁" in (body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", t1, {"orgIds": [org_a["id"]], "taskId": task_id, "excludeRecentHours": 0}).get("msg") or ""), "（再打一次仍应是 42901）") # 清掉计数，别影响后续/重复运行 os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning ' 'DEL xt:urge:task:%d > NUL 2>&1' % task_id) stok = mp(S1) b = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", stok, {"taskId": task_id}) check("学生不能催报 → 403", b.get("code") in (40101, 40301, 403), f"code={b.get('code')} msg={b.get('msg')}") b = body_of("POST", f"/api/v1/mp/teacher/notify/unsubmitted", None, {"taskId": task_id}) check("未登录催报 → 40101", b.get("code") == 40101, f"code={b.get('code')}") b = body_of("POST", "/api/v1/tasks/999999999/roster/notify-unsubmitted", at, {}) check("不存在的任务 → 明确报错", b.get("code") != 0, f"code={b.get('code')}") 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 _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"138001366{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: for tid in CREATED_TASK_IDS: R.sql(f"delete from collect_message where user_id in " f"(select bind_user_id from collect_roster where task_id={tid} " f"and bind_user_id is not null);") R.sql(f"delete from collect_notify_record where receiver_id in " f"(select bind_user_id from collect_roster where task_id={tid} " f"and bind_user_id is not null);") 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_org where task_id={tid};") R.sql(f"delete from collect_material_def where task_id={tid};") R.sql(f"delete from collect_task where id={tid};") R.sql("delete from collect_message where user_id in " "(993711, 993712, 993713);") R.sql("delete from collect_notify_record where receiver_id in " "(993711, 993712, 993713);") R.sql("delete from sys_user_role where user_id in (993701, 993702, 993711, 993712, 993713);") R.sql("delete from sys_user where id in (993701, 993702, 993711, 993712, 993713);") if os.path.exists(TMP_ROSTER): os.remove(TMP_ROSTER) print(" 已清理：任务 / 名册 / 填报单 / 消息 / 通知记录 / 测试账号 / 临时 Excel") if __name__ == "__main__": raise SystemExit(main())