xiaotutongxue/scripts/e2e-m3-draft.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
草稿保存 验收测试（《03-接口设计》§9.3 / §9.4）。

    [x] 取草稿是幂等的（反复调用只产生一份）
    [x] 保存草稿：只传变化的字段，服务端合并
    [x] **锁定字段服务端再拒一次**（前端只读是可以绕过的）
    [x] 改动留痕：只有"老师填过且学生改过"的字段进 changed_fields
    [x] 学生把预填字段**清空**也要能落库（updateById 会静默跳过 null）
    [x] 未过门禁不能建草稿；已提交的不能直接改
    [x] 草稿值会覆盖名册值回显，且"清空过的"字段不会被名册值顶回来

跑法：python scripts/e2e-m3-draft.py
"""
from __future__ import annotations

import argparse
import base64
import hashlib
import hmac
import json
import os
import re
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

sys.stdout.reconfigure(encoding="utf-8")

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEV_YML = os.path.join(ROOT, "backend", "xt-server", "src", "main", "resources", "application-dev.yml")
PG = os.environ.get("XT_PG_CONTAINER", "xt-pg")

PASSED: list[str] = []
FAILED: list[tuple[str, str]] = []
CREATED: list[int] = []


def ok(n: str) -> None:
    PASSED.append(n)
    print(f"  [OK]   {n}")


def fail(n: str, d: str) -> None:
    FAILED.append((n, d))
    print(f"  [FAIL] {n}\n         {d}")


def check(n: str, cond: bool, d: str = "") -> bool:
    if cond:
        ok(n)
    else:
        fail(n, d or "断言不成立")
    return cond


def sql(q: str) -> str:
    r = subprocess.run(
        ["docker", "exec", "-e", "PGPASSWORD=fill_dev_pwd", PG, "psql", "-U", "fill",
         "-d", "filldb", "-t", "-A", "-c", q], capture_output=True, text=True, timeout=40)
    return (r.stdout or r.stderr).strip()


def read_cfg() -> dict:
    files = [DEV_YML, os.path.join(os.path.dirname(DEV_YML), "application.yml")]
    text = "\n".join(open(f, encoding="utf-8").read() for f in files if os.path.exists(f))
    out = {}
    for k, p in [("jwt", r"^\s*secret:\s*(\S+)\s*$"), ("issuer", r"^\s*issuer:\s*(\S+)\s*$")]:
        m = re.search(p, text, re.M)
        if not m:
            raise SystemExit(f"找不到 {k}")
        out[k] = m.group(1)
    return out


def b64u(b: bytes) -> str:
    return base64.urlsafe_b64encode(b).rstrip(b"=").decode()


def mint(secret: str, issuer: str, uid: int, tenant: int, roles: list[str], utype: int) -> str:
    key = base64.b64decode(secret)
    now = int(time.time())
    signing = (b64u(json.dumps({"alg": "HS512"}, separators=(",", ":")).encode()) + "."
               + b64u(json.dumps({"iss": issuer, "sub": str(uid), "tenantId": tenant,
                                  "orgIds": [], "roles": roles, "userType": utype,
                                  "iat": now, "exp": now + 3600}, separators=(",", ":")).encode()))
    sig = hmac.new(key, signing.encode(), hashlib.sha512).digest()
    return signing + "." + b64u(sig)


def call(method: str, path: str, token: str | None = None, body: dict | None = None) -> dict:
    h = {"Content-Type": "application/json; charset=utf-8"}
    if token:
        h["Authorization"] = f"Bearer {token}"
    data = json.dumps(body, ensure_ascii=False).encode() if body is not None else None
    req = urllib.request.Request("http://localhost:8080" + path, data=data, headers=h, method=method)
    try:
        with urllib.request.urlopen(req, timeout=30) as x:
            return json.loads(x.read().decode("utf-8"))
    except urllib.error.HTTPError as e:
        try:
            return json.loads(e.read().decode("utf-8"))
        except Exception:
            return {"_http": e.code}


def fields_of(form: dict) -> dict:
    out = {}
    for sec in form.get("sections") or []:
        for f in sec.get("fields") or []:
            out[f["key"]] = f
    return out


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--task", type=int, default=17)
    ap.add_argument("--tenant", type=int, default=1)
    args = ap.parse_args()
    task_id = args.task
    tenant = args.tenant

    print("=" * 72)
    print(f"草稿保存 E2E　task={task_id}")
    print("=" * 72)

    cfg = read_cfg()
    uid = 991001  # 测试用学生（不建真实账号，草稿表只存 id）
    tok = mint(cfg["jwt"], cfg["issuer"], uid, tenant, ["STUDENT"], 4)
    base = f"/api/v1/mp"

    # ── 0. 前置：必须先过门禁 ──────────────────────────
    print("\n[0] 前置门槛")
    r = call("POST", f"{base}/submissions", tok, {"taskId": task_id})
    check("未过门禁时不能建草稿（40007）", r.get("code") == 40007,
          f"code={r.get('code')} msg={r.get('msg')}")

    # 造一个已绑定的名册行（模拟门禁已通过）
    h = sql(f"SELECT id FROM collect_roster WHERE task_id={task_id} LIMIT 1")
    if not h.isdigit():
        fail("取名册行", "任务下没有名册，先跑 e2e-m3-h5-auth 或手工造一行")
        return 1
    sql(f"UPDATE collect_roster SET bind_user_id={uid}, bind_type=1 WHERE id={h}")

    r = call("GET", f"{base}/tasks/{task_id}/form", tok)
    check("过门禁后能取表单", r.get("code") == 0, f"code={r.get('code')} msg={r.get('msg')}")
    form = r.get("data") or {}
    f0 = fields_of(form)
    check("表单有 21 个内置字段", len(f0) == 21, f"实际 {len(f0)}")
    check("报考职业是只读且已由老师填", f0["occupation"].get("locked") is True
          and f0["occupation"].get("value") == "车工", f"{f0['occupation']}")
    check("学历可改且标注老师已填", f0["education"].get("locked") is False
          and f0["education"].get("prefilled") is True, f"{f0['education']}")

    # ── 1. 取草稿（幂等）──────────────────────────────
    print("\n[1] 取草稿：幂等")
    r1 = call("POST", f"{base}/submissions", tok, {"taskId": task_id})
    check("首次创建", r1.get("code") == 0 and (r1.get("data") or {}).get("created") is True,
          f"{r1}")
    sid = (r1.get("data") or {}).get("id")
    CREATED.append(sid)
    r2 = call("POST", f"{base}/submissions", tok, {"taskId": task_id})
    check("再次调用返回同一份（created=false）",
          (r2.get("data") or {}).get("id") == sid and (r2.get("data") or {}).get("created") is False,
          f"{r2.get('data')}")
    cnt = sql(f"SELECT count(*) FROM collect_submission WHERE task_id={task_id} AND submitter_id={uid}")
    check("库里只有一份草稿", cnt == "1", f"{cnt}")

    # ── 2. 保存草稿 ──────────────────────────────────
    print("\n[2] 保存草稿（只传变化的字段）")
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {
        "school": "江西机电职业技术学院",
        "major": "数控加工",
        "enroll_date": "2024-09-01",
        "is_student": True,
    }})
    check("保存成功", r.get("code") == 0, f"code={r.get('code')} msg={r.get('msg')}")
    r = call("GET", f"{base}/tasks/{task_id}/form", tok)
    fd = fields_of(r.get("data") or {})
    check("存过的值回显了", fd["school"].get("value") == "江西机电职业技术学院",
          f"{fd['school'].get('value')}")
    check("库里也落了列（可统计）",
          sql(f"SELECT school FROM collect_submission WHERE id={sid}") == "江西机电职业技术学院",
          sql(f"SELECT school FROM collect_submission WHERE id={sid}"))
    check("没传的字段不受影响（名册值仍在）", fd["occupation"].get("value") == "车工",
          f"{fd['occupation'].get('value')}")
    check("草稿单 id 出现在表单里", (r.get("data") or {}).get("submissionId") == sid,
          f"{(r.get('data') or {}).get('submissionId')}")

    # ── 3. 锁定字段服务端再拒一次 ──────────────────────
    print("\n[3] 锁定字段：前端只读是可绕过的，服务端必须再拦")
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"occupation": "铣工"}})
    check("改「报考职业」被拒", r.get("code") == 40001, f"code={r.get('code')} msg={r.get('msg')}")
    check("拒绝理由讲清是老师定的、以及怎么办",
          "老师" in (r.get("msg") or "") and "班主任" in (r.get("msg") or ""),
          f"msg={r.get('msg')}")
    check("库里没被改", sql(f"SELECT occupation FROM collect_submission WHERE id={sid}") == "车工",
          sql(f"SELECT occupation FROM collect_submission WHERE id={sid}"))
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"job_level": "三级/高级工"}})
    check("改「报考等级」也被拒", r.get("code") == 40001, f"code={r.get('code')}")
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"occupation": "车工"}})
    check("把锁定字段改成原值不算改（不报错）", r.get("code") == 0, f"code={r.get('code')}")

    # ── 4. 未知字段要被拒绝 ────────────────────────────
    print("\n[4] 未知字段")
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"not_a_field": "x"}})
    check("传不认识的字段被拒（不静默忽略）", r.get("code") == 40001,
          f"code={r.get('code')} msg={r.get('msg')}")

    # ── 5. 改预填字段 → 留痕 ───────────────────────────
    print("\n[5] 改动留痕")
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"education": "大学专科"}})
    check("改预填的「学历」成功", r.get("code") == 0, f"code={r.get('code')}")
    cf = sql(f"SELECT changed_fields FROM collect_submission WHERE id={sid}")
    check("changed_fields 记录了 education", "education" in cf, f"changed_fields={cf}")
    check("没改过的字段不在里面", "occupation" not in cf and "school" not in cf,
          f"changed_fields={cf}")

    # ── 6. 清空预填字段（updateById 会静默跳过 null）────
    print("\n[6] 清空预填字段")
    r = call("PUT", f"{base}/submissions/{sid}", tok, {"formData": {"education": ""}})
    check("清空成功", r.get("code") == 0, f"code={r.get('code')}")
    v = sql(f"SELECT coalesce(education,'<NULL>') FROM collect_submission WHERE id={sid}")
    check("库里真的变成 NULL（不是静默没生效）", v == "<NULL>", f"education={v}")
    r = call("GET", f"{base}/tasks/{task_id}/form", tok)
    fd = fields_of(r.get("data") or {})
    check("清空后回显是空，**不会被名册值顶回来**", fd["education"].get("value") in (None, ""),
          f"{fd['education'].get('value')}")
    check("没动过的字段仍用名册值", fd["work_unit"].get("prefilled") in (True, False)
          and fd["occupation"].get("value") == "车工", f"{fd['occupation'].get('value')}")

    # ── 7. 清理 ───────────────────────────────────────
    print("\n[7] 清理")
    sql(f"DELETE FROM collect_submission WHERE id IN ({','.join(str(x) for x in CREATED)})")
    sql(f"UPDATE collect_roster SET bind_user_id=NULL, bind_time=NULL, bind_type=NULL WHERE id={h}")
    ok(f"已清理 {len(CREATED)} 份草稿并解绑名册")

    print("\n" + "=" * 72)
    print(f"通过 {len(PASSED)}　失败 {len(FAILED)}　共 {len(PASSED) + len(FAILED)}")
    for n, d in FAILED:
        print(f"  · {n}\n      {d}")
    print("=" * 72)
    return 1 if FAILED else 0


if __name__ == "__main__":
    raise SystemExit(main())