lovife
870c388287
学生端看到的表单终于可以按任务配置了。按计划表这一项是后端
（4 人日），表单设计器界面属 M1-07。
交付：
· GET/PUT /tasks/{id}/form-schema（§7.9 / §3.5）
· BuiltinFields：官方《考生线下报名信息导入模板》的 21 个内置字段（A~U 列）
—— 选项集逐字照抄，级联规则照官方原文
四条守住的规则：
1) 版本只增不改：每次保存插一行新版本，旧的保留（is_current=0）。
学生是按某一版填的，事后必须能回答"他当时看到的是什么"。表上没有 deleted
列也是这个道理 —— 表单版本不需要软删，它本来就不该删。
顺带：collect_form_schema 没有 deleted 列，实体刻意不继承基类
（collect_task_org 踩过同一个坑）。
2) baseVersion 是乐观锁，不符则 40025 并把最新定义一起回传。
设计器是"打开 → 改很多处 → 保存"，中间可能好几分钟；直接覆盖会静默丢掉
别人刚加的字段，而且丢失发生在"另一个人"那里。只说"版本冲突"而不给最新内容，
等于把合并推给一个信息不全的调用方 —— 所以 40025 的 data 里带着最新 schema。
3) 内置字段不可删（40026）的判断依据是「内置目录」而不是「上一版定义」：
拿上一版当依据，客户端只要让上一版先缺一个就绕过去了。E2E 也验了
把 builtin 传成 false 无法绕过 —— builtin 与 source 一律以内置目录为准。
4) 扩展字段键必须带 extra_ 前缀、不可与内置重名：扩展字段的值落 extras_json，
与内置列同处一份 form_data，一个叫 education 的扩展字段会让"这个值到底是
内置的还是扩展的"永远说不清。
契约之外必须补的一个字段 source：
§3.5 说"内置字段与 collect_submission 列名一致"，但「姓名」「身份证号」
「联系电话」「序号」根本不在 submission 里，在名册里（collect_roster）。
不标来源，填报引擎就得靠硬编码名单去猜"这个 key 去哪取值"，猜错的表现是
"学生端这两栏永远是空的"——一个不报错、只让人困惑的失败。
⚠️ source 与 locked 互相独立：「联系电话」source=ROSTER 但 locked=false
（预填可改，名册里常是家长的老号码）；「报考职业」source=SUBMISSION 且
locked=true（由计划定，学生不懂也不该改）。
21 项内置字段的口径：
· 选项集逐字照抄官方（学历 11 / 等级 5 / 补考理论 4 / 实操 6 / 综合 2）。
官方原文写着"若手动输入则只能填写下拉选项中包含的值"，多一个少一个都算改了规则。
· 级联：补考三科仅 is_retake=true 显示；在校生三项仅 is_student=true 显示；
入学时间带 ^\d{4}-\d{2}-\d{2}$ 正则（官方原文「格式必须 YYYY-MM-DD」）。
· 工作单位保留官方原文提示「没有工作单位请填「无」」。
⚠️「无」是合法业务值，不是空值，校验时不能把它当成"没填"拦掉。
· 锁定项（姓名/身份证号/报考职业/工种/等级）与 PRD 一致，另把「序号」一并锁定
（来自名册，让学生编辑"自己是第几号"没有意义）。
顺带：V6 更正 collect_form_schema.schema_json 的列注释（V1 写的是"字段定义数组"，
实际存 {"fields":[...],"extras":[...]}）。合成一个文档而不是拆两列，是因为
一版表单就是一份快照 —— 拆开就存在"只写成功一半"的可能，而"学生看到的那版"
必须是原子的。
本轮踩的坑（记进交接摘要）：
· 我的 E2E 又写错一条：测「已截止不可改表单」时先调 close，但**草稿是不能直接
截止的**（close 只允许 收集中/已暂停）。改成先用 SQL 夹具把发布前置备齐 →
发布 → 再截止。这反而验证了一件对的事：没配齐就发布，学生填了也没人审，
所以 close 把草稿挡住是合理的。
· 写 Java 时把 Map 声明成了 String，编译器直接拦住，没有代价。
已知缺口：
· 表单设计器界面未做（M1-07）。
· 报考职业/工种/等级的 options 为空：它们来自"评价计划"（由名册聚合，M2），
不是固定选项。属预期，等 M2 接上计划级联。
· 库里没有字典表，所以官方选项内联在 options 里，dictCode 暂未启用。
· 条件显示的求值引擎属 M3-10，本轮只做存取与形状校验。
验收：e2e-m1-form.py 48/48；六条 E2E 合计 326 项断言全通过；Flyway 迁移 1~6 success。
409 lines
21 KiB
Python
409 lines
21 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
M1-03 表单 schema 版本化与冲突检测 验收测试。
|
||
|
||
覆盖《03-接口设计》§7.9 / §3.5：
|
||
[x] 未保存过时返回内置字段目录的默认定义（官方 21 项，version=0）
|
||
[x] 22 项内置字段的锁定/必填/来源/选项与官方模板逐条对齐
|
||
[x] 保存为**新版本**（旧版本保留），is_current 只有一个
|
||
[x] baseVersion 不符 → 40025 **并回传最新定义**（前端据此合并）
|
||
[x] 缺少内置字段 → 40026（内置字段不可删）
|
||
[x] 扩展字段：必须 extra_ 前缀、不可与内置重名、不参与统计
|
||
[x] 客户端传 builtin=false 不能绕过「不可删除」（内建标志以目录为准）
|
||
[x] 状态限制与权限
|
||
|
||
为什么这些值得逐条断言：
|
||
表单定义是学生端看到的东西。版本搞错的表现是"学生按旧版填、老师按新版审"；
|
||
内置字段被删的表现是"这个人的姓名取不到值"—— 都是不报错、只让人困惑的失败。
|
||
|
||
用法
|
||
python scripts/e2e-m1-form.py
|
||
|
||
依赖：cryptography；且需要能 `docker exec xt-pg`（清理）
|
||
退出码：0 = 全部通过
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
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__)))
|
||
PUB_KEY = os.path.join(ROOT, "deploy", "rsa-public-key-dev.pem")
|
||
|
||
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]] = []
|
||
|
||
CTX: dict[str, str | None] = {"base": None, "token": None}
|
||
CREATED_TASK_IDS: list[int] = []
|
||
CREATED_USER_IDS: list[int] = []
|
||
|
||
|
||
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) -> tuple[bool, str]:
|
||
try:
|
||
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.returncode == 0, (r.stdout or r.stderr).strip()
|
||
except Exception as e:
|
||
return False, str(e)
|
||
|
||
|
||
def cleanup_all() -> None:
|
||
if CREATED_TASK_IDS:
|
||
ids = ",".join(str(i) for i in CREATED_TASK_IDS)
|
||
for stmt in [
|
||
f"DELETE FROM collect_form_schema WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_material_def WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_task_org WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_roster WHERE task_id IN ({ids})",
|
||
f"DELETE FROM collect_task WHERE id IN ({ids})",
|
||
]:
|
||
sql(stmt)
|
||
base, token = CTX.get("base"), CTX.get("token")
|
||
if base and token:
|
||
for uid in CREATED_USER_IDS:
|
||
try:
|
||
call("DELETE", f"{base}/api/v1/users/{uid}", token)
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def rsa_encrypt(plain: str) -> str:
|
||
from cryptography.hazmat.primitives import serialization
|
||
from cryptography.hazmat.primitives.asymmetric import padding
|
||
with open(PUB_KEY, "rb") as f:
|
||
pub = serialization.load_pem_public_key(f.read())
|
||
return base64.b64encode(pub.encrypt(plain.encode(), padding.PKCS1v15())).decode()
|
||
|
||
|
||
def call(method: str, url: str, token: str | None = None,
|
||
body: dict | None = None, timeout: int = 30) -> tuple[int, dict]:
|
||
headers = {"Content-Type": "application/json; charset=utf-8"}
|
||
if token:
|
||
headers["Authorization"] = f"Bearer {token}"
|
||
data = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
||
return resp.status, json.loads(resp.read().decode("utf-8"))
|
||
except urllib.error.HTTPError as e:
|
||
raw = e.read().decode("utf-8")
|
||
try:
|
||
return e.code, json.loads(raw)
|
||
except json.JSONDecodeError:
|
||
return e.code, {"raw": raw}
|
||
|
||
|
||
def login(base: str, username: str, password: str, tenant: str | None = None):
|
||
payload = {"username": username, "password": rsa_encrypt(password)}
|
||
if tenant:
|
||
payload["tenantCode"] = tenant
|
||
return call("POST", f"{base}/api/v1/auth/login", body=payload)
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="M1-03 表单 schema 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("=" * 72)
|
||
print(f"M1-03 表单 schema 版本化 E2E base={base} 租户={args.tenant}")
|
||
print("=" * 72)
|
||
|
||
suffix = uuid.uuid4().hex[:6]
|
||
|
||
# ── 0. 准备 ────────────────────────────────────────
|
||
print("\n[0] 准备")
|
||
status, body = login(base, "admin", args.password, args.tenant)
|
||
if body.get("code") != 0:
|
||
fail("管理员登录", f"HTTP {status} code={body.get('code')} msg={body.get('msg')}")
|
||
return 1
|
||
token = body["data"]["token"]
|
||
tenant_id = (body["data"].get("tenant") or {}).get("id")
|
||
CTX["base"], CTX["token"] = base, token
|
||
ok("管理员登录")
|
||
|
||
status, body = call("POST", f"{base}/api/v1/tasks", token, {
|
||
"taskName": f"E2E表单 {suffix}", "collectMode": 1, "batchNo": "2003",
|
||
"majorName": "数控加工", "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}）")
|
||
|
||
# ── 1. 默认定义 = 官方 21 项 ─────────────────────────
|
||
print("\n[1] 未保存过时的默认定义")
|
||
status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/form-schema", token)
|
||
d = body.get("data") or {}
|
||
fields = d.get("fields") or []
|
||
by_key = {f["key"]: f for f in fields}
|
||
check("返回内置默认定义（version=0，未落库）",
|
||
d.get("version") == 0 and d.get("latestVersion") == 0,
|
||
f"version={d.get('version')} latest={d.get('latestVersion')}")
|
||
check("恰好 21 个内置字段（官方模板 A~U 列）", len(fields) == 21, f"实际 {len(fields)}")
|
||
check("顺序即 A~U 列顺序",
|
||
[f["key"] for f in fields] == [
|
||
"seq_no", "real_name", "id_card_no", "education", "phone", "work_unit",
|
||
"occupation", "worktype", "job_level", "is_retake", "retake_theory",
|
||
"retake_practice", "retake_overall", "is_student", "school", "major",
|
||
"enroll_date", "work_years", "cert_occupation", "cert_level", "cert_no"],
|
||
f"{[f['key'] for f in fields]}")
|
||
check("全部标记为 builtin（内置不可删）", all(f.get("builtin") is True for f in fields),
|
||
f"{[(f['key'], f.get('builtin')) for f in fields if not f.get('builtin')]}")
|
||
|
||
# 锁定项：PRD §「锁定」= 姓名、身份证号、报考职业、报考工种、报考等级
|
||
locked_expected = {"real_name", "id_card_no", "occupation", "worktype", "job_level"}
|
||
locked_actual = {f["key"] for f in fields if f.get("locked")}
|
||
check("锁定项与 PRD 一致（姓名/身份证号/报考职业/工种/等级）",
|
||
locked_expected.issubset(locked_actual),
|
||
f"应为 {sorted(locked_expected)}，实际 {sorted(locked_actual)}")
|
||
check("联系电话来自名册但**不锁定**（预填可改：名册里常是家长的老号码）",
|
||
by_key["phone"].get("source") == "ROSTER" and by_key["phone"].get("locked") is False,
|
||
f"source={by_key['phone'].get('source')} locked={by_key['phone'].get('locked')}")
|
||
check("姓名/身份证号的来源是名册（不在 collect_submission 里）",
|
||
by_key["real_name"].get("source") == "ROSTER"
|
||
and by_key["id_card_no"].get("source") == "ROSTER",
|
||
f"{by_key['real_name'].get('source')} / {by_key['id_card_no'].get('source')}")
|
||
check("学历的来源是 submission 且带 11 个官方选项",
|
||
by_key["education"].get("source") == "SUBMISSION"
|
||
and len(by_key["education"].get("options") or []) == 11,
|
||
f"source={by_key['education'].get('source')} options={len(by_key['education'].get('options') or [])}")
|
||
check("报考等级的 5 个官方选项照抄",
|
||
(by_key["job_level"].get("options") or [])[:2] == ["五级/初级工", "四级/中级工"],
|
||
f"{by_key['job_level'].get('options')}")
|
||
check("补考三科只在「是否补考=是」时显示（官方规则 1）",
|
||
all((by_key[k].get("visibleWhen") or {}).get("field") == "is_retake"
|
||
and (by_key[k].get("visibleWhen") or {}).get("value") is True
|
||
for k in ("retake_theory", "retake_practice", "retake_overall")),
|
||
f"{[(k, by_key[k].get('visibleWhen')) for k in ('retake_theory', 'retake_practice')]}")
|
||
check("在校生三项只在「是否在校生=是」时显示（官方规则 3/4）",
|
||
all((by_key[k].get("visibleWhen") or {}).get("field") == "is_student"
|
||
for k in ("school", "major", "enroll_date")),
|
||
f"{[(k, by_key[k].get('visibleWhen')) for k in ('school', 'major', 'enroll_date')]}")
|
||
check("入学时间带 YYYY-MM-DD 正则（官方原文「格式必须 YYYY-MM-DD」）",
|
||
(by_key["enroll_date"].get("validate") or {}).get("regex") == "^\\d{4}-\\d{2}-\\d{2}$",
|
||
f"{by_key['enroll_date'].get('validate')}")
|
||
check("工作单位的提示语写着「没有则填无」", "无" in (by_key["work_unit"].get("tip") or ""),
|
||
f"{by_key['work_unit'].get('tip')}")
|
||
|
||
# ── 2. 保存为版本 1 ─────────────────────────────────
|
||
print("\n[2] 保存为新版本")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 0,
|
||
"fields": fields,
|
||
"extras": [{"key": "extra_skill_cert", "label": "其它技能证书编号",
|
||
"type": "text", "required": False, "sortNo": 100}],
|
||
})
|
||
check("从 version=0 保存成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}")
|
||
saved = body.get("data") or {}
|
||
check("新版本是 1", saved.get("version") == 1, f"{saved.get('version')}")
|
||
check("isCurrent=true", saved.get("isCurrent") is True, f"{saved.get('isCurrent')}")
|
||
check("扩展字段存下来了", len(saved.get("extras") or []) == 1,
|
||
f"{saved.get('extras')}")
|
||
check("latestVersion 同步为 1", saved.get("latestVersion") == 1, f"{saved.get('latestVersion')}")
|
||
|
||
okk, out = sql(f"SELECT count(*) FROM collect_form_schema WHERE task_id={task_id}")
|
||
check("库里落了一行版本", okk and out.strip() == "1", f"{out}")
|
||
|
||
status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/form-schema", token)
|
||
check("读回的是 version=1", (body.get("data") or {}).get("version") == 1,
|
||
f"{(body.get('data') or {}).get('version')}")
|
||
|
||
# ── 3. 改一处 → 版本 2 ──────────────────────────────
|
||
print("\n[3] 再次保存 → 版本递增、旧版保留")
|
||
fields2 = json.loads(json.dumps(fields))
|
||
for f in fields2:
|
||
if f["key"] == "work_unit":
|
||
f["required"] = False
|
||
f["tip"] = "改成选填了"
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 1, "fields": fields2, "extras": [],
|
||
})
|
||
check("从 version=1 保存成功", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}")
|
||
check("新版本是 2", (body.get("data") or {}).get("version") == 2, f"{(body.get('data') or {}).get('version')}")
|
||
okk, out = sql(f"SELECT count(*) FROM collect_form_schema WHERE task_id={task_id}")
|
||
check("库里两行版本（旧版保留，不覆盖）", okk and out.strip() == "2", f"{out}")
|
||
okk, out = sql(f"SELECT count(*) FROM collect_form_schema WHERE task_id={task_id} AND is_current=1")
|
||
check("当前版本只有一个", okk and out.strip() == "1", f"{out}")
|
||
status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/form-schema", token)
|
||
d2 = {f["key"]: f for f in (body.get("data") or {}).get("fields") or []}
|
||
check("改动已生效（工作单位变选填）", d2["work_unit"].get("required") is False,
|
||
f"{d2['work_unit'].get('required')}")
|
||
check("扩展字段可以整体清空", (body.get("data") or {}).get("extras") == [],
|
||
f"{(body.get('data') or {}).get('extras')}")
|
||
|
||
# ── 4. 版本冲突 ─────────────────────────────────────
|
||
print("\n[4] 版本冲突（40025）")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 1, "fields": fields2, "extras": [],
|
||
})
|
||
check("用过期 baseVersion 保存被拒 40025", body.get("code") == 40025,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
latest = body.get("data") or {}
|
||
check("40025 的响应里**带上了最新定义**（前端据此做差异合并）",
|
||
latest.get("version") == 2 and len(latest.get("fields") or []) == 21,
|
||
f"version={latest.get('version')} fields={len(latest.get('fields') or [])}")
|
||
check("冲突提示里说清了基准版本与最新版本",
|
||
"基准版本 1" in (body.get("msg") or "") and "最新 2" in (body.get("msg") or ""),
|
||
f"msg={body.get('msg')}")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 99, "fields": fields2, "extras": [],
|
||
})
|
||
check("baseVersion 比最新大也被拒（不是只防「小于」）", body.get("code") == 40025,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
|
||
# ── 5. 内置字段不可删 ───────────────────────────────
|
||
print("\n[5] 内置字段不可删（40026）")
|
||
missing_one = [f for f in fields2 if f["key"] != "id_card_no"]
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 2, "fields": missing_one, "extras": [],
|
||
})
|
||
check("删掉「身份证号」被拒 40026", body.get("code") == 40026,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
check("拒绝理由点出了缺哪个字段，并给出替代做法（隐藏而不是删）",
|
||
"id_card_no" in (body.get("msg") or "") and "隐藏" in (body.get("msg") or ""),
|
||
f"msg={body.get('msg')}")
|
||
|
||
spoofed = json.loads(json.dumps(fields2))
|
||
spoofed = [f for f in spoofed if f["key"] != "id_card_no"]
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 2, "fields": spoofed, "extras": [],
|
||
})
|
||
check("把 builtin 传成 false 也无法绕过不可删除", body.get("code") == 40026,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
|
||
# ── 6. 扩展字段 ─────────────────────────────────────
|
||
print("\n[6] 扩展字段")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 2, "fields": fields2,
|
||
"extras": [{"key": "education", "label": "想冒充内置列", "type": "text"}],
|
||
})
|
||
check("扩展字段不能与内置字段重名", body.get("code") == 40001,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 2, "fields": fields2,
|
||
"extras": [{"key": "skillCert", "label": "没有 extra_ 前缀", "type": "text"}],
|
||
})
|
||
check("扩展字段 key 必须带 extra_ 前缀", body.get("code") == 40001,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
check("以上失败都没有产生新版本",
|
||
(call("GET", f"{base}/api/v1/tasks/{task_id}/form-schema", token)[1].get("data") or {}).get("version") == 2,
|
||
"版本被推进了")
|
||
|
||
# ── 7. 状态限制与权限 ───────────────────────────────
|
||
print("\n[7] 状态限制与权限")
|
||
# ⚠️ 要测「已截止不可改表单」，必须先把任务推到「收集中」：
|
||
# 草稿是**不能直接截止的**（close 只允许 收集中 / 已暂停）——
|
||
# 这正是发布前置校验存在的意义：没配齐就发布，学生填了也没人审。
|
||
# 所以这里用 SQL 夹具把发布前置的三样备齐。
|
||
sql(f"INSERT INTO collect_roster (tenant_id, task_id, real_name, id_card_no, id_card_hash) "
|
||
f"VALUES ({tenant_id}, {task_id}, 'E2E学生A', 'seed', 'seed')")
|
||
sql(f"INSERT INTO collect_material_def (tenant_id, task_id, material_code, material_name, required) "
|
||
f"VALUES ({tenant_id}, {task_id}, 'IDCARD', '身份证', 1)")
|
||
okk, out = sql(f"INSERT INTO collect_task_org (tenant_id, task_id, org_id, status) "
|
||
f"VALUES ({tenant_id}, {task_id}, "
|
||
f"(SELECT id FROM sys_org WHERE tenant_id={tenant_id} AND org_type=3 LIMIT 1), 1)")
|
||
check("夹具：名册 / 材料 / 覆盖班级就位", okk, out)
|
||
|
||
status, body = call("POST", f"{base}/api/v1/tasks/{task_id}/publish", token, {"force": True})
|
||
check("发布任务（进入收集中）", body.get("code") == 0,
|
||
f"code={body.get('code')} msg={body.get('msg')}")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 2, "fields": fields2, "extras": [],
|
||
})
|
||
check("收集中允许改表单（§3.5 说「已发布任务改表单必须递增」，说明是允许改的）",
|
||
body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}")
|
||
|
||
status, body = call("POST", f"{base}/api/v1/tasks/{task_id}/close", token, {})
|
||
check("把任务截止", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", token, {
|
||
"baseVersion": 3, "fields": fields2, "extras": [],
|
||
})
|
||
check("已截止不可再改表单", body.get("code") == 40001, f"code={body.get('code')} msg={body.get('msg')}")
|
||
status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/form-schema", token)
|
||
check("已截止仍可读表单（只读不等于看不见）", body.get("code") == 0, f"code={body.get('code')}")
|
||
|
||
ht_user = f"htf{suffix}"
|
||
status, body = call("POST", f"{base}/api/v1/users", token, {
|
||
"username": ht_user, "realName": f"E2E表单班主任{suffix}",
|
||
"phone": "13900139401", "roleCodes": ["HEAD_TEACHER"], "userType": 3,
|
||
})
|
||
if check("造班主任账号", body.get("code") == 0, f"code={body.get('code')} msg={body.get('msg')}"):
|
||
uid, pwd = body["data"]["id"], body["data"]["initialPassword"]
|
||
CREATED_USER_IDS.append(uid)
|
||
status, body = login(base, ht_user, pwd, args.tenant)
|
||
ht_token = body["data"]["token"]
|
||
status, body = call("GET", f"{base}/api/v1/tasks/{task_id}/form-schema", ht_token)
|
||
check("班主任能读表单定义", body.get("code") == 0, f"code={body.get('code')}")
|
||
status, body = call("PUT", f"{base}/api/v1/tasks/{task_id}/form-schema", ht_token, {
|
||
"baseVersion": 2, "fields": fields2, "extras": [],
|
||
})
|
||
check("班主任不能改表单（无 task:manage）", body.get("code") == 40301,
|
||
f"code={body.get('code')}")
|
||
|
||
# ── 8. 清理 ────────────────────────────────────────
|
||
print("\n[8] 清理测试数据")
|
||
cleanup_all()
|
||
ok(f"已清理 {len(CREATED_TASK_IDS)} 个任务、{len(CREATED_USER_IDS)} 个账号")
|
||
|
||
print("\n" + "=" * 72)
|
||
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
|
||
if FAILED:
|
||
print("\n失败项：")
|
||
for name, detail in FAILED:
|
||
print(f" · {name}\n {detail}")
|
||
print("=" * 72)
|
||
return 1 if FAILED else 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
code = main()
|
||
except BaseException:
|
||
print("\n ⚠ 脚本异常退出，正在清理已造出的测试数据…")
|
||
cleanup_all()
|
||
raise
|
||
raise SystemExit(code)
|