xiaotutongxue/scripts/e2e-m2-scope.py

# -*- 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)