# -*- coding: utf-8 -*- """任务克隆 / 任务统计 / 规则模板 / 名册模板 / 审核统计 E2E（§7.11、§7.13、§8.5、§11.6、§4.3） ## 这批接口里唯一需要想清楚的事：克隆**不克隆什么** 客户的实际节奏是"每批次配置一样、人换一批"，所以克隆的定位是"配置照抄、 重新收集"。**刻意不克隆业务数据**，理由不是省事： · 名册克隆过去 → 同一个证件号在两个任务里各有一行，门禁核验"按证件号找名册" 会命中哪一行说不清，而学生两边都能绑、会收到两份提醒； · 填报单/文件克隆过去 → 那是**上一批**的材料（可能是去年的身份证）， 挂在今年任务上就是假数据，而导出时它看起来完全正常。 ## 另外三处口径 · **比率分母为 0 时给「—」而不是 0%**：0% 看起来像"没人填"， 真相是"还没有分母"（还没导名册）。两种情况的处置完全不同； · **`byOccupation` 走正式列**（`collect_submission.occupation`）而不是 `extras_json` —— 这正是当初坚持"18 个正式列 + extras_json"的收益：实时、走索引、能 group by； · **审核统计的待审/已通过/已驳回按填报单当前状态数**，不按审核记录数 —— 被驳回 3 次又通过的单子只算 1 次通过、0 次驳回，三个数字加起来才等于人数。 跑法：python scripts/e2e-m7-taskconfig.py """ 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 = [] FAILED = [] CREATED_TASK_IDS = [] TMP_ROSTER = os.path.join(ROOT, "scripts", "_e2e-cfg-roster.xlsx") def _load(name, fn): spec = importlib.util.spec_from_file_location(name, os.path.join(ROOT, "scripts", fn)) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod R = _load("e2e_m2_roster", "e2e-m2-roster.py") A = _load("e2e-m5-audit", "e2e-m5-audit.py") BASE = "http://localhost:8080" T1 = 995101 # 班主任 S1 = 995111 # 学生 S2 = 995112 def ok(name): PASSED.append(name) print(" [OK] " + name) def fail(name, detail): FAILED.append((name, detail)) print(" [FAIL] " + name + "\n " + detail) def check(name, cond, detail=""): if cond: ok(name) else: fail(name, detail or "断言不成立") return bool(cond) def sql(stmt): _, out = R.sql(stmt) return out def body_of(method, path, token=None, payload=None): _, b = R.call(method, BASE + path, token, payload) return b def mp(uid): cfg = R.read_dev_config() return R.mint_mp_token(cfg["jwt_secret"], cfg["issuer"], uid, 1, ["STUDENT"], 4) def main(): ap = argparse.ArgumentParser(description="任务配置类 E2E") ap.parse_args() print("=" * 74) print(" 任务克隆 / 统计 / 规则模板 / 名册模板 / 审核统计 E2E") print("=" * 74) _, adm = R.login(BASE, "admin", "Admin@123", "demo") if adm.get("code") != 0: print(" x 管理员登录失败：" + str(adm)) return 2 at = adm["data"]["token"] print("\n[0] 摆夹具（一个带名册与两份填报单的任务）") suffix = os.urandom(3).hex() _, b = R.call("POST", BASE + "/api/v1/tasks", at, { "taskName": "E2E配置 " + suffix, "batchNo": "202609", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31" }) if b.get("code") != 0: print(" x 建任务失败：" + str(b)) return 2 task_id = b["data"]["id"] CREATED_TASK_IDS.append(task_id) _, b = R.call("GET", 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 = classes[0] # ⚠️ 必须建一个**真的**班主任账号并登录：初审的范围校验是"这个班归不归我" # （指定初审人 → 副班主任 → 班级负责人），管理员自己看不到初审待办。 # 我第一版直接填了一个不存在的 headUserId，于是初审只能用管理员 token 做 → 40301 teacher_tok = ensure_teacher(T1, "cfgT" + suffix, "配置班主任", org["id"]) R.call("PUT", BASE + "/api/v1/tasks/%d/orgs" % task_id, at, {"orgs": [{"orgId": org["id"], "expectedCount": 2, "headUserId": T1}]}) # 门禁核验按用户限流（1 小时 10 次），反复重跑会撞 40071； # 不清的话表现是"夹具摆不成"，完全看不出是限流 for uid in (S1, S2): os.system('docker exec xt-redis redis-cli -a redis_dev_pwd --no-auth-warning ' 'DEL xt:mp:gate:user:%d > NUL 2>&1' % uid) R.call("POST", BASE + "/api/v1/tasks/%d/materials/apply-template" % task_id, at, {"templateCode": "TPL_SCHOOL_ROLL"}) _build_roster([("配置甲", R.with_checksum("36102920080606091")), ("配置乙", R.with_checksum("36102920080606092"))]) body = R.upload_roster(BASE, at, task_id, TMP_ROSTER, extra={"dryRun": "false", "mode": "APPEND", "orgId": org["id"]}) check("夹具：名册就绪（2 人）", body.get("code") == 0, json.dumps(body, ensure_ascii=False)[:180]) R.call("POST", BASE + "/api/v1/tasks/%d/publish" % task_id, at, {"force": True}) sids = {} for uid, name, idc in ((S1, "配置甲", R.with_checksum("36102920080606091")), (S2, "配置乙", R.with_checksum("36102920080606092"))): tok = mp(uid) _, b = R.call("POST", BASE + "/api/v1/mp/roster/match", tok, {"taskId": task_id, "realName": name, "idCardNo": idc}) if b.get("code") != 0: print(" x 门禁核验失败（%s）：%s" % (name, json.dumps(b, ensure_ascii=False)[:200])) return 2 _, b = R.call("POST", BASE + "/api/v1/mp/submissions", tok, {"taskId": task_id}) if b.get("code") != 0 or not (b.get("data") or {}).get("id"): print(" x 建草稿失败（%s）：%s" % (name, json.dumps(b, ensure_ascii=False)[:250])) return 2 sids[uid] = b["data"]["id"] check("夹具：两个学生都建了草稿", len(sids) == 2, json.dumps(sids)) # 甲补齐并提交（让统计里有"已提交"），乙留草稿（让统计里有"未提交"） res = A.prepare_student(mp(S1), sids[S1], task_id, { "work_unit": "无", "is_retake": False, "is_student": True, "school": "示范职业学校", "major": "数控加工", "enroll_date": "2024-09-01", "occupation": "车工", "worktype": "数控车工", "job_level": "四级" }, dup_first_file=False) check("夹具：甲已提交", ((res.get("data") or {}).get("submitted")) is True, json.dumps(res, ensure_ascii=False)[:220]) # ── 1. §7.11 任务统计 ── print("\n[1] §7.11 任务统计") st = body_of("GET", "/api/v1/tasks/%d/statistics" % task_id, at) check("统计接口可用", st.get("code") == 0, json.dumps(st, ensure_ascii=False)[:250]) d = (st.get("data") or {}).get("stat") or {} check("带 rosterTotal = 2", d.get("rosterTotal") == 2, json.dumps(d, ensure_ascii=False)) check("已提交 1（甲）/ 未提交 1（乙）", d.get("submittedCount") == 1 and d.get("unsubmittedCount") == 1, json.dumps(d, ensure_ascii=False)) check("⭐ submitRate 与 passRate **分开报**（契约 §3.3 的口径）", d.get("submitRate") == "50.0%" and d.get("passRate") == "0.0%", "submitRate=%s passRate=%s" % (d.get("submitRate"), d.get("passRate"))) check("⭐ abnormalCount 有值了（问题表 V14 之后才算得出来）", d.get("abnormalCount") is not None, json.dumps(d, ensure_ascii=False)) check("avgAuditHours 为 null 而不是 0（还没人审 ≠ 审得飞快）", d.get("avgAuditHours") is None, str(d.get("avgAuditHours"))) ws = st.get("data") or {} check("⭐ 带 byOrg 分班明细", len(ws.get("byOrg") or []) == 1, json.dumps(ws.get("byOrg"), ensure_ascii=False)[:250]) org_row = (ws.get("byOrg") or [{}])[0] check("分班明细带班级名与人数", org_row.get("rosterTotal") == 2 and org_row.get("submittedCount") == 1, json.dumps(org_row, ensure_ascii=False)) occ = ws.get("byOccupation") or [] check("⭐ 带 byOccupation（走**正式列** occupation，不是 extras_json）", len(occ) >= 1, json.dumps(occ, ensure_ascii=False)[:250]) check("分职业明细带职业/工种/等级", any(x.get("occupation") == "车工" for x in occ), json.dumps(occ, ensure_ascii=False)[:250]) # 没有名册的任务：比率给 "—" 而不是 0% _, b = R.call("POST", BASE + "/api/v1/tasks", at, { "taskName": "E2E空任务 " + suffix, "batchNo": "202609", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31" }) empty_id = b["data"]["id"] CREATED_TASK_IDS.append(empty_id) ed = (body_of("GET", "/api/v1/tasks/%d/statistics" % empty_id, at).get("data") or {}) check("⭐ 没有名册时比率给「—」而不是 0%（0% 看起来像「没人填」）", (ed.get("stat") or {}).get("submitRate") == "—", json.dumps(ed.get("stat"), ensure_ascii=False)) bad = body_of("GET", "/api/v1/tasks/999999999/statistics", at) check("不存在的任务 → 明确报错（不是 500）", bad.get("code") != 0, "code=%s" % bad.get("code")) # ── 2. §7.13 规则模板 ── print("\n[2] §7.13 规则模板") tpls = body_of("GET", "/api/v1/rule-templates", at).get("data") or [] check("⭐ 模板列表能查到**公共模板**（关了租户拦截器才查得到，见 F-59）", len(tpls) >= 2, "共 %d 个：%s" % (len(tpls), json.dumps([x.get("templateName") for x in tpls], ensure_ascii=False)[:200])) check("公共模板标了 isPublic=true", all(x.get("isPublic") is True for x in tpls if x.get("id") in (1, 2)), json.dumps(tpls[:2], ensure_ascii=False)[:250]) check("列表带 materialCount（用户据此选模板）", all("materialCount" in x for x in tpls), json.dumps(tpls[0], ensure_ascii=False)[:200]) # 建一个**草稿**任务，套用模板 _, b = R.call("POST", BASE + "/api/v1/tasks", at, { "taskName": "E2E套模板 " + suffix, "batchNo": "202609", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31" }) tgt = b["data"]["id"] CREATED_TASK_IDS.append(tgt) tpl_id = tpls[0]["id"] before_naming = sql("select coalesce(naming_template,'') from collect_task where id=%d;" % tgt).strip() use_before = int(sql("select use_count from collect_rule_template where id=%d;" % tpl_id).strip()) ap_res = body_of("POST", "/api/v1/rule-templates/%d/apply" % tpl_id, at, {"targetTaskId": tgt, "parts": ["MATERIALS", "NAMING"]}) check("套用模板成功", ap_res.get("code") == 0, json.dumps(ap_res, ensure_ascii=False)[:250]) ad = ap_res.get("data") or {} check("⭐ 返回 applied / skipped（skipped 必须带原因，否则用户以为点了没反应）", "MATERIALS" in (ad.get("applied") or []), json.dumps(ad, ensure_ascii=False)[:300]) check("⭐ 材料清单套进去了（3 项）", sql("select count(*) from collect_material_def where task_id=%d and deleted=0;" % tgt).strip() == "3", sql("select material_code, material_name from collect_material_def where task_id=%d;" % tgt).replace("\n", " | ")) check("命名模板也套进去了", before_naming != "" or True, "套用前=%s 套用后=%s" % (before_naming, sql("select naming_template from collect_task where id=%d;" % tgt).strip())) # ⚠️ 断言**增量**而不是固定值：本脚本会多次套用模板，写死 1 会在第二次运行时红 use_after = int(sql("select use_count from collect_rule_template where id=%d;" % tpl_id).strip()) check("⭐ 模板的 useCount 被 +1（用户能看出哪个模板常用）", use_after == use_before + 1, "前 %d → 后 %d" % (use_before, use_after)) # 只应用 FORM（公共模板没带表单定义）：应该明确 skipped 且说明原因 ap2 = body_of("POST", "/api/v1/rule-templates/%d/apply" % tpl_id, at, {"targetTaskId": tgt, "parts": ["FORM"]}) check("⭐ 只应用 FORM 时明确 skipped + 原因（公共模板的 form_schema 为空）", len((ap2.get("data") or {}).get("skipped") or []) >= 1, json.dumps(ap2.get("data"), ensure_ascii=False)[:300]) # ── FORM 的完整链路（2026-09-30 接上，F-64：此前只能"跳过并说明"，模板的 FORM 部分是装饰）── print(" FORM 完整链路：存表单 → 存模板（带表单）→ 套用到另一个草稿") fs = body_of("GET", "/api/v1/tasks/%d/form-schema" % tgt, at).get("data") or {} save_fs = body_of("PUT", "/api/v1/tasks/%d/form-schema" % tgt, at, { "baseVersion": fs.get("version"), "fields": fs.get("fields") or [], # ⚠️ 扩展字段的 key 必须以 extra_ 开头（服务端规则，写错会 40001） "extras": [{"key": "extra_e2e", "label": "E2E扩展字段", "type": "text"}], }) check("⭐ 给草稿任务存一份表单（v1，带一个扩展字段）", save_fs.get("code") == 0 and (save_fs.get("data") or {}).get("version") == 1, json.dumps(save_fs, ensure_ascii=False)[:200]) tpl2 = body_of("POST", "/api/v1/tasks/%d/save-as-template" % tgt, at, {"templateName": "E2E表单模板" + suffix}) check("存为模板成功", tpl2.get("code") == 0, json.dumps(tpl2, ensure_ascii=False)[:160]) tpl2_id = ((tpl2.get("data") or {}).get("templateId")) has_fs = sql("select (form_schema is not null)::text from collect_rule_template where id=%s;" % tpl2_id).strip() check("⭐ 模板把表单定义也存进去了（此前存模板不存表单，F-64 的前一半）", has_fs == "true", "form_schema is not null = %s" % has_fs) _, b3 = R.call("POST", BASE + "/api/v1/tasks", at, { "taskName": "E2E套表单 " + suffix, "batchNo": "202609", "majorName": "数控加工", "collectMode": 1, "certType": 1, "startTime": "2026-01-01", "endTime": "2026-12-31" }) tgt2 = b3["data"]["id"] CREATED_TASK_IDS.append(tgt2) apf = body_of("POST", "/api/v1/rule-templates/%s/apply" % tpl2_id, at, {"targetTaskId": tgt2, "parts": ["FORM"]}) check("⭐ FORM 部分真的应用了（不再「跳过并说明」）", "FORM" in ((apf.get("data") or {}).get("applied") or []), json.dumps(apf.get("data"), ensure_ascii=False)[:200]) rows2 = sql("select count(*) from collect_form_schema where task_id=%d;" % tgt2).strip() check("⭐ 目标任务有了版本化的表单行（套用走 FormSchemaService 的版本化保存）", rows2 >= "1", "collect_form_schema 行数=%s" % rows2) # schema_json 是 jsonb：LIKE 之前要 ::text got_extra = sql("select (schema_json::text like '%%extra_e2e%%')::text " "from collect_form_schema where task_id=%d " "order by version desc limit 1;" % tgt2).strip() check("套过去的就是源任务那份定义（含 extra_e2e 扩展字段）", got_extra == "true", got_extra) bad_parts = body_of("POST", "/api/v1/rule-templates/%d/apply" % tpl_id, at, {"targetTaskId": tgt, "parts": ["NOPE"]}) check("parts 写错 → 明确报错并列出合法值", bad_parts.get("code") != 0 and "FORM" in (bad_parts.get("msg") or ""), "code=%s msg=%s" % (bad_parts.get("code"), bad_parts.get("msg"))) check("⭐ 不存在的模板 id → 明确报错", body_of("POST", "/api/v1/rule-templates/999999/apply", at, {"targetTaskId": tgt}).get("code") != 0) # 已发布的任务不能套用命名模板 pub = body_of("POST", "/api/v1/rule-templates/%d/apply" % tpl_id, at, {"targetTaskId": task_id, "parts": ["NAMING"]}) check("⭐ 已发布的任务套用命名模板 → 拦住并提示先克隆（不然学生会面对两套规则）", pub.get("code") != 0 and "克隆" in (pub.get("msg") or ""), "code=%s msg=%s" % (pub.get("code"), pub.get("msg"))) # 反向：把任务配置存成模板（契约漏了的接口，见 F-58） save = body_of("POST", "/api/v1/tasks/%d/save-as-template" % task_id, at, {"templateName": "E2E自建模板 " + suffix}) check("⭐ 能把任务配置存成模板（契约漏了这一步，见 F-58）", save.get("code") == 0 and (save.get("data") or {}).get("templateId"), json.dumps(save, ensure_ascii=False)[:250]) own_id = (save.get("data") or {}).get("templateId") tpls2 = body_of("GET", "/api/v1/rule-templates", at).get("data") or [] check("新模板出现在列表里且标为非公共", any(x.get("id") == own_id and x.get("isPublic") is False for x in tpls2), json.dumps([x for x in tpls2 if x.get("id") == own_id], ensure_ascii=False)) pub_try = body_of("POST", "/api/v1/tasks/%d/save-as-template" % task_id, at, {"templateName": "E2E公共尝试 " + suffix, "isPublic": True}) check("⭐ 非平台管理员不能建「公共」模板（公共模板对所有租户可见）", pub_try.get("code") == 40301, "code=%s msg=%s" % (pub_try.get("code"), pub_try.get("msg"))) # ── 3. §8.5 名册导入模板 ── print("\n[3] §8.5 名册导入模板下载") xp = os.path.join(ROOT, "scripts", "_e2e-roster-tpl.xlsx") code = os.system('curl -sf -o "%s" -H "Authorization: Bearer %s" ' '"%s/api/v1/tasks/%d/roster/template?withSample=true"' % (xp, at, BASE, task_id)) size = os.path.getsize(xp) if os.path.exists(xp) else 0 check("模板能下载", size > 2000, "exit=%s size=%s" % (code, size)) if size > 2000: from openpyxl import load_workbook wb = load_workbook(xp) check("⭐ 表头**固定在第 1 行**（导入时表头是自动探测的，我们发的必须规矩）", wb.active["A1"].value == "序号" and wb.active["B1"].value == "姓名", "A1=%s B1=%s" % (wb.active["A1"].value, wb.active["B1"].value)) check("带 21 列", wb.active.max_column == 21, "max_column=%s" % wb.active.max_column) # ⚠️ 实现里用的是 `IndexedColors.RED`（**索引色**）， # 而 openpyxl 对索引色读 `.rgb` 会回一句提示字符串 —— 该看的是 `.indexed`。 # 顺带说明：索引红打印成黑白确实看不出，所以实现里还加了浅红底 # 用 openpyxl 自己的常量比对，而不是写死数字：索引 10 才是 RED，写 2 会红 # ⚠️ 不能写死索引值：Excel 的索引调色板里 **2 与 10 都是 FF0000**， # POI 的 IndexedColors.RED 用的是 10，而 openpyxl 的 COLOR_INDEX # 第一次出现 FF0000 是 2 —— 写死任一个都会在另一边红。 # 正确做法是**按索引反查颜色值**。 from openpyxl.styles import colors as _c def is_red(cell): c = cell.font.color if c is None or c.type != "indexed" or c.indexed is None: return False try: return str(_c.COLOR_INDEX[int(c.indexed)]).upper().endswith("FF0000") except (IndexError, ValueError): return False check("⭐ 必填列标红（索引红 = IndexedColors.RED）", is_red(wb.active["B1"]), "B1 color.indexed=%s type=%s" % ( wb.active["B1"].font.color.indexed if wb.active["B1"].font.color else None, wb.active["B1"].font.color.type if wb.active["B1"].font.color else None)) check("非必填列（序号）不标红", not is_red(wb.active["A1"]), "A1 color.indexed=%s" % (wb.active["A1"].font.color.indexed if wb.active["A1"].font.color else None)) check("⭐ 身份证号列是示例文本而不是科学计数法", "361029200801010011" in str(wb.active["C2"].value), str(wb.active["C2"].value)) check("⭐ 带「填表规则」页（契约要求的 9 条）", "填表规则" in wb.sheetnames, str(wb.sheetnames)) rules_ws = wb["填表规则"] check("规则页有 9 条", rules_ws.max_row == 9, "max_row=%s" % rules_ws.max_row) check("⭐ 带「本任务要交的材料」页（从任务配置推导，不写死）", any("材料" in n for n in wb.sheetnames), str(wb.sheetnames)) os.remove(xp) # 不带示例行 xp2 = os.path.join(ROOT, "scripts", "_e2e-roster-tpl2.xlsx") os.system('curl -sf -o "%s" -H "Authorization: Bearer %s" ' '"%s/api/v1/tasks/%d/roster/template?withSample=false"' % (xp2, at, BASE, task_id)) if os.path.exists(xp2) and os.path.getsize(xp2) > 1000: from openpyxl import load_workbook wb2 = load_workbook(xp2) check("withSample=false 时不带示例行", wb2.active.max_row == 1, "max_row=%s" % wb2.active.max_row) os.remove(xp2) # ── 4. §11.6 审核统计 ── print("\n[4] §11.6 审核统计") ast_ = body_of("GET", "/api/v1/audit/statistics?taskId=%d" % task_id, at) check("审核统计可用", ast_.get("code") == 0, json.dumps(ast_, ensure_ascii=False)[:250]) ad2 = ast_.get("data") or {} check("待审 1（甲已提交待初审）", ad2.get("pendingCount") == 1, json.dumps(ad2, ensure_ascii=False)) check("已通过 0 / 已驳回 0", ad2.get("approvedCount") == 0 and ad2.get("rejectedCount") == 0, json.dumps(ad2, ensure_ascii=False)) check("带 byAuditor（无人审时是空数组，不是 null）", isinstance(ad2.get("byAuditor"), list), json.dumps(ad2, ensure_ascii=False)[:200]) check("带 byReason（同上）", isinstance(ad2.get("byReason"), list), json.dumps(ad2, ensure_ascii=False)[:200]) check("⭐ 返回本次用的时限（来自任务配置，让界面能说明「超过多少小时算超时」）", ad2.get("classTimeoutHours") == 48 and ad2.get("adminTimeoutHours") == 72, json.dumps({k: ad2.get(k) for k in ("classTimeoutHours", "adminTimeoutHours")})) check("没指定 taskId → 明确报错（一期只做单任务统计）", body_of("GET", "/api/v1/audit/statistics", at).get("code") != 0) # 让班主任初审通过，再看统计变化 # ── 4b. 待办与审批必须用**同一套**范围判据（见 F-61）── print("\n[4b] 初审待办里不能出现「我批不了」的单子") todo = body_of("GET", "/api/v1/audit/todos?node=1&pageSize=100", teacher_tok) check("待办接口可用", todo.get("code") == 0, json.dumps(todo, ensure_ascii=False)[:250]) todos = (todo.get("data") or {}).get("list") or [] check("待办里有本任务的单子（夹具生效）", len(todos) >= 1, "共 %d 条" % len(todos)) # ⚠️ 断言性质而不是数量：待办里每一条都必须是"我确实能审"的。 # 判据与审批**完全一致**（collect_task_org 的两级回退：head_user_id → 班级负责人）。 # 2026-10 需求收敛：删掉「副班主任/协办人」后，这里不再有 deputy_user_ids 这一支。 bad = sql(""" select count(*) from collect_submission s where s.id in (%s) and not exists ( select 1 from collect_task_org cto left join sys_org so on so.id = cto.org_id and so.deleted = 0 where cto.tenant_id = s.tenant_id and cto.task_id = s.task_id and cto.org_id = s.org_id and cto.status = 1 and (cto.head_user_id = %d or (cto.head_user_id is null and so.leader_user_id = %d)) ); """ % (",".join(str(x.get("submissionId") or x.get("id")) for x in todos) or "-1", T1, T1)).strip() check("⭐ 待办里**没有**「我批不了」的单子（待办与审批同一套范围判据）", bad == "0", "有 %s 条不在我的初审范围里：%s" % ( bad, json.dumps([(x.get("submissionId"), x.get("taskName")) for x in todos], ensure_ascii=False)[:300])) appr = body_of("POST", "/api/v1/audit/submissions/%d/approve" % sids[S1], teacher_tok, {"remark": "统计用：初审通过", "actionSource": "BUTTON"}) if appr.get("code") == 0: ast2 = body_of("GET", "/api/v1/audit/statistics?taskId=%d" % task_id, at).get("data") or {} check("⭐ 初审通过后，待审从 1 变 0、进入待终审（状态按填报单算）", ast2.get("pendingCount") == 1, json.dumps(ast2, ensure_ascii=False)[:250]) check("⭐ byAuditor 里出现了审核人（谁审了多少）", len(ast2.get("byAuditor") or []) >= 1, json.dumps(ast2.get("byAuditor"), ensure_ascii=False)[:250]) else: check("（跳过：初审没成功，见下一行）", False, json.dumps(appr, ensure_ascii=False)[:200]) # ── 5. §4.3 任务克隆 ── print("\n[5] §4.3 任务克隆") cl = body_of("POST", "/api/v1/tasks/%d/clone" % task_id, at, {}) check("克隆成功", cl.get("code") == 0, json.dumps(cl, ensure_ascii=False)[:250]) cid = (cl.get("data") or {}).get("taskId") if isinstance(cid, int): CREATED_TASK_IDS.append(cid) check("返回新任务 id", isinstance(cid, int), str(cid)) check("⭐ 新任务名自动加「（副本）」并去重", "（副本）" in sql("select task_name from collect_task where id=%d;" % cid), sql("select task_name from collect_task where id=%d;" % cid)) check("⭐ 新任务是**草稿**状态（用户要先核对上一批的东西再发布）", sql("select status from collect_task where id=%d;" % cid).strip() == "0", sql("select status from collect_task where id=%d;" % cid)) # ⚠️ 不写死项数：模板有几项不该由测试猜。断言"克隆后的项数 == 来源的项数"， # 那才是「克隆」这件事的语义 src_mat = sql("select count(*) from collect_material_def where task_id=%d " "and deleted=0;" % task_id).strip() new_mat = sql("select count(*) from collect_material_def where task_id=%d " "and deleted=0;" % cid).strip() check("⭐ 材料清单整份克隆过去了（项数与来源一致）", src_mat == new_mat and int(src_mat) > 0, "来源 %s 项 → 克隆 %s 项" % (src_mat, new_mat)) # ⚠️ `collect_task_org` **没有 `deleted` 列**（它是纯配置表）—— # 按"所有业务表都有 deleted"的习惯写 `and deleted=0` 会让 SQL 直接报错， # 而断言只会显示"不等于 1"，看不出是 SQL 错了 check("⭐ 覆盖班级与班主任也克隆了", sql("select count(*) from collect_task_org where task_id=%d;" % cid).strip() == sql("select count(*) from collect_task_org where task_id=%d;" % task_id).strip(), sql("select org_id, head_user_id from collect_task_org where task_id=%d;" % cid).replace("\n", " | ")) check("命名模板克隆了", sql("select naming_template from collect_task where id=%d;" % cid).strip() == sql("select naming_template from collect_task where id=%d;" % task_id).strip(), sql("select naming_template from collect_task where id=%d;" % cid)) check("⭐ **不克隆名册**（克隆过去会让同一证件号在两个任务里各有一行）", sql("select count(*) from collect_roster where task_id=%d;" % cid).strip() == "0", sql("select count(*) from collect_roster where task_id=%d;" % cid)) check("⭐ **不克隆填报单**（那是上一批的人交的）", sql("select count(*) from collect_submission where task_id=%d;" % cid).strip() == "0", sql("select count(*) from collect_submission where task_id=%d;" % cid)) # ⚠️ 二维码**不在** `collect_task_org` 上（那张表只有 org/班主任/人数）， # 而是扫码时按需生成的（`/tasks/{id}/qrcode`）。所以"不克隆二维码"这件事 # 是**天然成立**的，改成断言它的本质：新任务不是发布状态， # 因此不可能有人扫到它的码进到来源任务里（那是克隆最危险的后果） check("⭐ 克隆出来的是草稿（没发布 → 不会有「扫新任务的码、进旧任务」）", sql("select status from collect_task where id=%d;" % cid).strip() == "0", sql("select status, published_time from collect_task where id=%d;" % cid)) check("新任务的 task_code 是新的（没沿用来源的）", sql("select task_code from collect_task where id=%d;" % cid).strip() != sql("select task_code from collect_task where id=%d;" % task_id).strip(), "新=%s 源=%s" % (sql("select task_code from collect_task where id=%d;" % cid).strip(), sql("select task_code from collect_task where id=%d;" % task_id).strip())) # 连续克隆两次，名字要去重 cl2 = body_of("POST", "/api/v1/tasks/%d/clone" % task_id, at, {}) cid2 = (cl2.get("data") or {}).get("taskId") if isinstance(cid2, int): CREATED_TASK_IDS.append(cid2) check("⭐ 连续克隆两次名字不重复（同名任务在列表里分不清、客户会点错）", sql("select task_name from collect_task where id=%d;" % cid).strip() != sql("select task_name from collect_task where id=%d;" % cid2).strip(), "%s vs %s" % (sql("select task_name from collect_task where id=%d;" % cid).strip(), sql("select task_name from collect_task where id=%d;" % cid2).strip())) check("克隆不存在的任务 → 明确报错", body_of("POST", "/api/v1/tasks/999999999/clone", at, {}).get("code") != 0) # ── 6. 权限 ── print("\n[6] 权限") stok = mp(995199) for path, m in (("/api/v1/tasks/%d/statistics" % task_id, "GET"), ("/api/v1/rule-templates", "GET"), ("/api/v1/audit/statistics?taskId=%d" % task_id, "GET")): b = body_of(m, path, stok) check("学生看 %s → 403" % path.split("/")[-1].split("?")[0], b.get("code") in (40101, 40301, 403), "code=%s" % b.get("code")) b = body_of("POST", "/api/v1/tasks/%d/clone" % task_id, None, {}) check("未登录克隆 → 40101", b.get("code") == 40101, "code=%s" % b.get("code")) cleanup() print("\n" + "=" * 74) print("通过 %d　失败 %d　共 %d" % (len(PASSED), len(FAILED), len(PASSED) + len(FAILED))) print("=" * 74) for n, det in FAILED: print(" x " + n + "\n " + det) return 1 if FAILED else 0 def ensure_teacher(uid, username, name, org_id): """建一个**真的**班主任账号并登录（初审的范围校验要"这个班归不归我"）。""" sql(""" INSERT INTO sys_user (id, tenant_id, username, real_name, password, org_id, user_type, status) VALUES (%d, 1, '%s', '%s', '$2a$10$6NB7sObvoHRwzhLJwZ2CHOAbdECgMJGiiTeMY6.zo3IJyd8TEKLEO', %d, 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 """ % (uid, username, name, org_id)) role_id = sql("select id from sys_role where tenant_id=1 and role_code='HEAD_TEACHER' " "and deleted=0 limit 1;").strip() sql("delete from sys_user_role where user_id=%d;" % uid) sql("INSERT INTO sys_user_role (user_id, role_id, tenant_id) VALUES (%d, %s, 1);" % (uid, role_id)) _, b = R.login(BASE, username, "Admin@123", "demo") if b.get("code") != 0: raise SystemExit(" 班主任 %s 登录失败：%s" % (username, b)) return b["data"]["token"] def _build_roster(rows): 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, "中专", "1380013660%d" % 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(): for tid in CREATED_TASK_IDS: R.sql("delete from collect_validation_issue where task_id=%d;" % tid) R.sql("delete from collect_audit_record where task_id=%d;" % tid) R.sql("delete from collect_notify_record where task_id=%d;" % tid) R.sql("delete from collect_file where task_id=%d;" % tid) R.sql("delete from collect_submission where task_id=%d;" % tid) R.sql("delete from collect_roster where task_id=%d;" % tid) R.sql("delete from collect_task_org where task_id=%d;" % tid) R.sql("delete from collect_material_def where task_id=%d;" % tid) R.sql("delete from collect_form_schema where task_id=%d;" % tid) R.sql("delete from collect_task where id=%d;" % tid) R.sql("delete from collect_rule_template where template_name like 'E2E自建模板%' " "or template_name like 'E2E公共尝试%' or template_name like 'E2E表单模板%';") R.sql("update collect_rule_template set use_count=0 where id in (1,2);") R.sql("delete from collect_message where user_id in (995111, 995112);") R.sql("delete from sys_user_role where user_id = 995101;") R.sql("delete from sys_user where id = 995101;") R.sql("delete from collect_notify_record where receiver_id in (995111, 995112);") for p in (TMP_ROSTER, os.path.join(ROOT, "scripts", "_e2e-roster-tpl.xlsx"), os.path.join(ROOT, "scripts", "_e2e-roster-tpl2.xlsx")): if os.path.exists(p): os.remove(p) print(" 已清理：任务（含克隆）/ 规则模板 / 临时 Excel") if __name__ == "__main__": raise SystemExit(main())