# -*- coding: utf-8 -*- """ 报考范围 E2E（M2-04 / M2-05，《03-接口设计》§7.7 / §7.8，PRD §5.3.1）—— FR-307/308/309/311。 验的核心是四件事： ① **聚合**：不存在"计划清单文件"，范围 = 名册里「职业 × 工种 × 等级」的去重组合。 所以要能按组合分组、给出每组人数与**是哪几个人**（`rosterIds`）。 ② **一致性体检**：40 人里 38 人报 A、2 人报 B，那 2 个很可能是填错 —— 而这种错**逐行核是发现不了的**（真实批次几百人）。所以必须机器算出来， 并且指出是**哪几个人**。只说"存在不一致"等于没说。 ③ **确认时的全覆盖约束**（这条是这一步真正的价值）： `rosterIds` 的并集必须恰好等于名册全员，差集返回 40024 并**点名**。 漏一个人，导出包就会缺人，而打开导出包时**不会有人数出 27 个**。 ④ **FR-311 追加告警**：范围确认之后，中途补录的人若报了范围外的报考项， 导入体检报告里必须立刻出现提示。 跑法：python scripts/e2e-m2-scope.py 前提：后端 8080 在跑、xt-pg 容器在跑、装了 openpyxl。 """ from __future__ import annotations import argparse import importlib.util import io import json import os import subprocess import sys import urllib.error import urllib.request import uuid if hasattr(sys.stdout, "reconfigure"): 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") PG_USER = os.environ.get("PG_USER", "fill") PG_DB = os.environ.get("PG_DB", "filldb") PG_PWD = os.environ.get("PG_PASSWORD", "fill_dev_pwd") PASSED: list[str] = [] FAILED: list[tuple[str, str]] = [] CREATED_TASK_IDS: list[int] = [] TMP_FILES: list[str] = [] # ── 复用名册 E2E 的工具（登录取 token、multipart 上传、校验位）───── def load_roster_e2e(): """ 直接 import e2e-m2-roster.py，避免夹具与工具出现两份 —— 两份就会漂移，而"夹具漂移"会让两个脚本一个过一个不过，且看不出谁对。 """ path = os.path.join(ROOT, "scripts", "e2e-m2-roster.py") spec = importlib.util.spec_from_file_location("e2e_m2_roster", path) mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod R = load_roster_e2e() 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", f"PGPASSWORD={PG_PWD}", PG_CONTAINER, "psql", "-U", PG_USER, "-d", PG_DB, "-t", "-A", "-c", stmt], capture_output=True, text=True, timeout=40) return (r.stdout or r.stderr).strip() def cleanup_all() -> None: if CREATED_TASK_IDS: ids = ",".join(str(i) for i in CREATED_TASK_IDS) sql(f"DELETE FROM collect_roster WHERE task_id IN ({ids})") sql(f"DELETE FROM collect_material_def WHERE task_id IN ({ids})") sql(f"DELETE FROM collect_task_org WHERE task_id IN ({ids})") sql(f"DELETE FROM collect_submission WHERE task_id IN ({ids})") sql(f"DELETE FROM collect_task WHERE id IN ({ids})") for p in TMP_FILES: try: os.remove(p) except OSError: pass # ── 夹具：一份故意做出"不一致"的名单 ─────────────────── # (姓名, 职业, 工种, 等级) —— 刻意构造出四类体检问题 ROWS = [ ("曾伊豪", "车工", "数控车工", "四级"), ("杨晨", "车工", "数控车工", "四级"), ("赵六", "车工", "数控车工", "四级"), ("钱七", "车工", "数控车工", "四级"), ("孙八", "车工", "数控车工", "四级"), ("周九", "车工", "数控车工", "四级"), ("吴十", "车工", "数控车工", "**三级**"), # ← 同职业多等级（少数派） ("郑十一", "铣工", "数控铣工", "四级"), # ← 任务范围外的职业（任务声明的是"车工"） ("王十二", "", "", ""), # ← 漏填报考职业 ] # 9 行 → 三个分组：车工/数控车工/四级(6人)、车工/数控车工/三级(1人)、铣工/数控铣工/四级(1人)、空(1人) GROUP_COUNT = 4 def build_fixture(path: 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, occ, wt, lv) in enumerate(ROWS): # ⚠️ 前 17 位必须是「6 位地区 + 8 位真实出生日期 + 3 位顺序」—— # 出生日期不存在（比如 0000）会被判格式非法，9 行全被阻断， # 而导入接口仍然回 code=0（它只回体检报告），看着像"导进去了"。 idc = R.with_checksum(f"361029200501{i + 1:02d}{i:03d}") row = i + 2 vals = [i + 1, name, idc, "中专", f"1380013800{i}", "无", occ, wt, lv, "否", "", "", "", "是", "示范职业学校", "数控加工", "2024-09-01"] for c, v in enumerate(vals, start=1): if v != "": ws.cell(row=row, column=c, value=v) wb.save(path) def build_extra_fixture(path: str, occupation: str, worktype: str, level: str, name: str) -> None: """追加一个人（用来验 FR-311：范围外的报考项要告警）。""" 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) idc = R.with_checksum("361029" + "19990101" + "001") vals = [1, name, idc, "中专", "13800138999", "无", occupation, worktype, level, "否", "", "", "", "是", "示范职业学校", "数控加工", "2024-09-01"] for c, v in enumerate(vals, start=1): if v != "": ws.cell(row=2, column=c, value=v) wb.save(path) # ── 主流程 ─────────────────────────────────────────── 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("/") print("=" * 76) print(f"报考范围 E2E（M2-04/05）　base={base}") print("=" * 76) suffix = uuid.uuid4().hex[:6] # ── 0. 准备 ──────────────────────────────────────── print("\n[0] 准备") _, body = R.login(base, "admin", args.password, args.tenant) if body.get("code") != 0: fail("管理员登录", f"code={body.get('code')} msg={body.get('msg')}") return 1 admin = body["data"]["token"] ok("管理员登录") # jobName 声明为「车工」：这样"铣工"那行才会被判成范围外（§7.7 的 SCOPE_OUT_OF_TASK） _, body = R.call("POST", f"{base}/api/v1/tasks", admin, { "taskName": f"E2E报考范围 {suffix}", "collectMode": 1, "batchNo": "2006", "majorName": "数控加工", "jobName": "车工", "certType": 1, "startTime": "2026-09-28", "endTime": "2026-12-31", }) if body.get("code") != 0: fail("建任务", f"code={body.get('code')} msg={body.get('msg')}") return 1 task_id = body["data"]["id"] CREATED_TASK_IDS.append(task_id) ok(f"建任务（id={task_id}，声明职业=车工）") fix = os.path.join(ROOT, "scripts", f"_tmp_scope_{suffix}.xlsx") TMP_FILES.append(fix) build_fixture(fix) body = R.upload_roster(base, admin, task_id, fix, {"dryRun": "false"}) check("名册导入成功（9 人）", body.get("code") == 0, f"{body}") # 断言"新增 9 行"而不只是 code=0：导入接口对"9 行全被阻断"也是回 code=0， # 只判 code 会把"一个人都没进去"当成功（这次就是这么被抓出来的） check("导入报告：新增 9 行、0 阻断", (body.get("data") or {}).get("created") == 9 and (body.get("data") or {}).get("failed") == 0, f"created={(body.get('data') or {}).get('created')} " f"failed={(body.get('data') or {}).get('failed')} " f"errors={((body.get('data') or {}).get('errors') or [])[:2]}") _, cnt = ("", sql(f"SELECT count(*) FROM collect_roster WHERE task_id={task_id} AND deleted=0")) check("库里 9 行", cnt == "9", f"count={cnt}") # ── 1. 聚合 ──────────────────────────────────────── print("\n[1] 聚合：范围 = 名册里「职业 × 工种 × 等级」的去重组合") _, g = R.call("GET", f"{base}/api/v1/tasks/{task_id}/plan-scope", admin) d = g.get("data") or {} check("读取成功", g.get("code") == 0, f"{g}") check("source = ROSTER_AGGREGATED（不存在从文件导入的路径）", d.get("source") == "ROSTER_AGGREGATED", f"{d.get('source')}") check("rosterCount = 9（分母）", d.get("rosterCount") == 9, f"{d.get('rosterCount')}") check("自动聚合成 4 个分组", len(d.get("groups") or []) == GROUP_COUNT, f"{[(x.get('occupation'), x.get('worktype'), x.get('jobLevel'), x.get('count')) for x in (d.get('groups') or [])]}") check("未确认（confirmed=false）", d.get("confirmed") is False, f"{d.get('confirmed')}") groups = d.get("groups") or [] main_g = next((x for x in groups if x.get("occupation") == "车工" and x.get("jobLevel") == "四级"), None) check("主分组：车工/数控车工/四级 = 6 人", main_g and main_g.get("count") == 6, f"{main_g}") check("分组带 rosterIds（要说得出是哪几个人）", main_g and len(main_g.get("rosterIds") or []) == 6, f"{main_g and main_g.get('rosterIds')}") # ── 2. 一致性体检 ────────────────────────────────── print("\n[2] 一致性体检：必须指出**是哪几个人**") hc = d.get("healthCheck") or {} check("passed = false（有未处理的问题）", hc.get("passed") is False, f"{hc.get('passed')}") issues = hc.get("issues") or [] by_code = {i.get("code"): i for i in issues} check("问题码齐全（漏填 / 同职业多等级 / 范围外职业）", {"SCOPE_UNFILLED", "SCOPE_LEVEL_MIXED", "SCOPE_OUT_OF_TASK"} <= set(by_code), f"{sorted(by_code)}") check("SCOPE_UNFILLED：1 人漏填报考职业", by_code.get("SCOPE_UNFILLED", {}).get("rosterIds") and len(by_code["SCOPE_UNFILLED"]["rosterIds"]) == 1, f"{by_code.get('SCOPE_UNFILLED')}") check("SCOPE_UNFILLED 带人话说明（含「无法导出」的后果）", "导出" in (by_code.get("SCOPE_UNFILLED", {}).get("detail") or ""), f"{by_code.get('SCOPE_UNFILLED', {}).get('detail')}") check("SCOPE_LEVEL_MIXED：指出**少数派**那 1 人（不是把 7 个人都算异常）", len(by_code.get("SCOPE_LEVEL_MIXED", {}).get("rosterIds") or []) == 1, f"{by_code.get('SCOPE_LEVEL_MIXED')}") check("SCOPE_LEVEL_MIXED 的说明里点了等级与人数", "四级" in (by_code.get("SCOPE_LEVEL_MIXED", {}).get("detail") or ""), f"{by_code.get('SCOPE_LEVEL_MIXED', {}).get('detail')}") check("SCOPE_OUT_OF_TASK：铣工那 1 人（任务声明的是车工）", len(by_code.get("SCOPE_OUT_OF_TASK", {}).get("rosterIds") or []) == 1, f"{by_code.get('SCOPE_OUT_OF_TASK')}") check("有问题的那几组被标成 NEED_REVIEW", any(x.get("status") == "NEED_REVIEW" for x in groups), f"{[(x.get('occupation'), x.get('status')) for x in groups]}") # ── 3. ★ 全覆盖约束：漏一个人必须被点名 ────────────── print("\n[3] 确认时的全覆盖约束（这一步真正的价值）") full = [{"occupation": x["occupation"], "worktype": x["worktype"], "jobLevel": x["jobLevel"], "status": "CONFIRMED", "rosterIds": x["rosterIds"]} for x in groups] missing_one = [dict(x, rosterIds=x["rosterIds"][:-1]) if i == 0 else x for i, x in enumerate(full)] _, r1 = R.call("PUT", f"{base}/api/v1/tasks/{task_id}/plan-scope", admin, {"groups": missing_one}) check("**漏一个人 → 40024**", r1.get("code") == 40024, f"{r1}") check("并且**点名**漏了谁（不能只说校验失败）", "人没有被分到" in (r1.get("msg") or ""), f"{r1.get('msg')}") check("还说明了后果（导出包少人发现不了）", "导出包" in (r1.get("msg") or ""), f"{r1.get('msg')}") dup = [dict(full[0], rosterIds=full[0]["rosterIds"] + full[1]["rosterIds"])] + full[1:] _, r2 = R.call("PUT", f"{base}/api/v1/tasks/{task_id}/plan-scope", admin, {"groups": dup}) check("同一个人被分到两组 → 40024", r2.get("code") == 40024, f"{r2}") check("提示里说明「一个人只能在一个组」", "多个组" in (r2.get("msg") or ""), f"{r2.get('msg')}") _, r3 = R.call("PUT", f"{base}/api/v1/tasks/{task_id}/plan-scope", admin, {"groups": full + [{"occupation": "焊工", "worktype": "无", "jobLevel": "五级", "status": "CONFIRMED", "rosterIds": [999999999]}]}) check("分组里带了不属于本任务的名册项 → 明确报错", r3.get("code") != 0, f"{r3}") # ── 4. 确认（含排除与忽略）────────────────────────── print("\n[4] 确认与修正（FR-309）") unfilled_g = next(x for x in groups if not x.get("occupation")) excluded = [dict(x, rosterIds=x["rosterIds"], status="CONFIRMED") for x in groups if x["occupation"]] excluded.append({"occupation": unfilled_g["occupation"], "worktype": unfilled_g["worktype"], "jobLevel": unfilled_g["jobLevel"], "status": "EXCLUDED", "rosterIds": unfilled_g["rosterIds"]}) _, r4 = R.call("PUT", f"{base}/api/v1/tasks/{task_id}/plan-scope", admin, {"groups": excluded, "remark": "漏填报考项的那位本轮先排除"}) check("全员覆盖 → 确认成功", r4.get("code") == 0, f"{r4}") d4 = r4.get("data") or {} check("confirmed = true", d4.get("confirmed") is True, f"{d4.get('confirmed')}") check("被排除的那组状态保留为 EXCLUDED", any(x.get("status") == "EXCLUDED" for x in (d4.get("groups") or [])), f"{[(x.get('occupation'), x.get('status')) for x in (d4.get('groups') or [])]}") check("**没忽略问题时 passed 仍为 false**（确认 ≠ 掩饰）", (d4.get("healthCheck") or {}).get("passed") is False, f"{(d4.get('healthCheck') or {}).get('passed')}") ignored = ["SCOPE_UNFILLED", "SCOPE_LEVEL_MIXED", "SCOPE_OUT_OF_TASK", "SCOPE_TYPEO_MISMATCH"] _, r5 = R.call("PUT", f"{base}/api/v1/tasks/{task_id}/plan-scope", admin, {"groups": excluded, "ignoredIssueCodes": ignored, "remark": "已知悉，本轮按此执行"}) d5 = r5.get("data") or {} check("显式忽略后 passed = true", (d5.get("healthCheck") or {}).get("passed") is True, f"{d5.get('healthCheck')}") check("被忽略的问题不再出现在待办里", (d5.get("healthCheck") or {}).get("issues") == [], f"{(d5.get('healthCheck') or {}).get('issues')}") check("updatedAt 有值（确认时间留痕）", d5.get("updatedAt") is not None, f"{d5.get('updatedAt')}") _, row = ("", sql(f"SELECT plan_scope_confirmed || '|' || coalesce(plan_scope_json::text,'-') " f"FROM collect_task WHERE id={task_id}")) check("库里 plan_scope_confirmed = 1", row.split("|")[0] == "1", f"{row[:80]}") check("范围快照落库（plan_scope_json 非空且含分组）", "rosterIds" in row and "EXCLUDED" in row, f"{row[:120]}") # ── 5. FR-311：追加名单的范围外告警 ────────────────── print("\n[5] FR-311：范围确认后，追加名单出现范围外的报考项要告警") extra_bad = os.path.join(ROOT, "scripts", f"_tmp_extra_bad_{suffix}.xlsx") TMP_FILES.append(extra_bad) build_extra_fixture(extra_bad, "焊工", "电焊工", "五级", "临时新人") rb = R.upload_roster(base, admin, task_id, extra_bad, {"dryRun": "false"}) codes_b = [w.get("code") for w in ((rb.get("data") or {}).get("warnings") or [])] check("范围外的报考项 → 导入体检里出现 SCOPE_OUT_OF_TASK 警告", "SCOPE_OUT_OF_TASK" in codes_b, f"{codes_b}") warn = next((w for w in ((rb.get("data") or {}).get("warnings") or []) if w.get("code") == "SCOPE_OUT_OF_TASK"), {}) check("告警说明了该怎么做（有意增补 / 名单拿错）", "有意增补" in (warn.get("reason") or ""), f"{warn.get('reason')}") extra_ok = os.path.join(ROOT, "scripts", f"_tmp_extra_ok_{suffix}.xlsx") TMP_FILES.append(extra_ok) build_extra_fixture(extra_ok, "车工", "数控车工", "四级", "临时新人二") rg = R.upload_roster(base, admin, task_id, extra_ok, {"dryRun": "false"}) codes_g = [w.get("code") for w in ((rg.get("data") or {}).get("warnings") or [])] check("范围**内**的报考项不告警（不能全报）", "SCOPE_OUT_OF_TASK" not in codes_g, f"{codes_g}") print() print("=" * 76) print(f"通过 {len(PASSED)}　失败 {len(FAILED)}　共 {len(PASSED) + len(FAILED)}") if FAILED: for n, det in FAILED: print(f" · {n}\n {det}") print("=" * 76) return 1 if FAILED else 0 if __name__ == "__main__": try: code = main() except SystemExit: raise except Exception as e: # noqa: BLE001 print(f"\n ✗ 脚本异常：{type(e).__name__}: {e}") code = 1 finally: cleanup_all() print(" 已清理：名册 / 任务 / 临时 Excel") raise SystemExit(code)