lovife
30c3714d35
新增：check-api-parity（契约 121 接口 ↔ 控制器，自带「必须恰好 121」自检）、 check-entity-columns（基类要求的列 vs 迁移建表列，F-35 那个坑踩了四次之后上的工具）、 check-manual（手册引用的错误码与提示文案对得上代码）、check-test-profile（test profile 真起一次）、 e2e-prodguard、load-test（四场景、按端点分别统计）、verify-all（一条命令跑完交付自检）、 dev-seed-audit（界面验收用的演示数据），以及 M1~M9 各轮 E2E。 压测抓到一个真缺陷并修掉（F-92）：门禁 IP 限流原来计「所有请求」，于是一个班连同一个 WiFi 时 60 个不同学生同时核验只过了 20 个 → 改为只计失败 + 单独的成功上限，20/60 → 60/60。 度量工具的偏差会伪装成产品缺陷（F-60/F-77/F-97），所以每个脚本都断言「动作自己的结论字段」 （submitted / created / canExport），而不是 HTTP 200 + code 0。 验证：verify-all.py --quick 5/5；--with-e2e 见 docs/09 §8。
311 lines
15 KiB
Python
311 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""字典管理 E2E（《03-接口设计》§13.3）。
|
||
|
||
这一屏的验收点不在于"能不能增删改"，而在于**三条边界的口径**：
|
||
|
||
① **内置字典的项名改不了**（只可停用单项 + 追加自定义项）——
|
||
内置选项照抄官方表格，改名会让导出的汇总表与官方模板对不上，
|
||
而那时没有任何地方会报错；
|
||
② **项码建后不可改** —— 历史数据里存的是项码，改了它过去填过的值
|
||
会指向一个不存在的项，而查询**不报错、只显示空白**；
|
||
③ **有下级项的项不可停用**（40036）—— 否则会出现"选了父项却选不到子项"，
|
||
而数据层面看不出矛盾。
|
||
|
||
另外两条容易做错的：
|
||
· `itemCount` 只算**启用中**的项（把停用的也算进去，管理员会以为界面漏了）；
|
||
· 内置项的"停用"与自定义项的"停用"是**两回事**，分开存才能"一键恢复默认"。
|
||
|
||
跑法：python scripts/e2e-m1-dict.py
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import importlib.util
|
||
import json
|
||
import os
|
||
import sys
|
||
import urllib.parse
|
||
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
PASSED: list[str] = []
|
||
FAILED: list[tuple[str, str]] = []
|
||
|
||
|
||
def _load(module_name: str, filename: str):
|
||
path = os.path.join(ROOT, "scripts", filename)
|
||
spec = importlib.util.spec_from_file_location(module_name, path)
|
||
mod = importlib.util.module_from_spec(spec)
|
||
spec.loader.exec_module(mod)
|
||
return mod
|
||
|
||
|
||
R = _load("e2e_m2_roster", "e2e-m2-roster.py")
|
||
BASE = "http://localhost:8080"
|
||
|
||
# 测试用字典码（用自定义字典，不污染内置的）
|
||
TEST_DICT = "E2E_SUBJECT"
|
||
TENANT = 1
|
||
|
||
|
||
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:
|
||
_, out = R.sql(stmt)
|
||
return out
|
||
|
||
|
||
def body_of(method: str, path: str, token: str | None = None, payload: dict | None = None) -> dict:
|
||
_, b = R.call(method, f"{BASE}{path}", token, payload)
|
||
return b
|
||
|
||
|
||
def dict_of(token: str) -> dict:
|
||
d = body_of("GET", "/api/v1/dicts", token).get("data") or []
|
||
return {x["dictCode"]: x for x in d}
|
||
|
||
|
||
def items_of(token: str, code: str) -> dict:
|
||
return body_of("GET", f"/api/v1/dicts/{code}/items", token).get("data") or {}
|
||
|
||
|
||
def main() -> int:
|
||
ap = argparse.ArgumentParser(description="字典管理 E2E")
|
||
ap.parse_args()
|
||
|
||
print("=" * 74)
|
||
print(" 字典管理 E2E（§13.3）")
|
||
print("=" * 74)
|
||
|
||
_, adm = R.login(BASE, "admin", "Admin@123", "demo")
|
||
if adm.get("code") != 0:
|
||
print(f" ✗ 管理员登录失败：{adm}")
|
||
return 2
|
||
at = adm["data"]["token"]
|
||
cleanup()
|
||
|
||
# ── 1. 内置字典目录 ──
|
||
print("\n[1] 内置字典目录（6 个）")
|
||
ds = dict_of(at)
|
||
check("字典列表可用", len(ds) >= 6, f"{sorted(ds)}")
|
||
expect = {"EDU_LEVEL": 11, "JOB_LEVEL": 5, "CERT_TYPE": 3,
|
||
"RETARGET_SUBJECT": 3, "REJECT_REASON": 10}
|
||
for code, n in expect.items():
|
||
check(f"{code} 项数 = {n}", (ds.get(code) or {}).get("itemCount") == n,
|
||
json.dumps(ds.get(code), ensure_ascii=False))
|
||
check("⭐ 内置字典都标了 builtin=true",
|
||
all((ds.get(c) or {}).get("builtin") is True for c in expect),
|
||
json.dumps({c: (ds.get(c) or {}).get("builtin") for c in expect}))
|
||
check("OCCUPATION 也在列表里（国标职业库，一期为空时 0 项）",
|
||
"OCCUPATION" in ds, f"{sorted(ds)}")
|
||
|
||
# ── 2. 学历项与表单口径一致 ──
|
||
print("\n[2] 学历项必须与表单下拉一致（两个入口看同一份口径）")
|
||
edu = items_of(at, "EDU_LEVEL")
|
||
names = [i["itemName"] for i in edu.get("list") or []]
|
||
check("学历 11 项", len(names) == 11, f"{names}")
|
||
check("含官方里那几项（照抄不得擅改）",
|
||
all(x in names for x in ("博士研究生", "技工学校", "其他")), f"{names}")
|
||
check("⭐ 内置项没有 id（它们在代码里，没有数据库行）",
|
||
all(i.get("id") is None for i in edu.get("list") or []),
|
||
json.dumps((edu.get("list") or [])[:1], ensure_ascii=False))
|
||
check("内置项都标了 builtin=true",
|
||
all(i.get("builtin") is True for i in edu.get("list") or []))
|
||
# 与表单口径对一遍：这是"两个入口一份口径"的机器化验证
|
||
jl = items_of(at, "JOB_LEVEL")
|
||
jnames = [i["itemName"] for i in jl.get("list") or []]
|
||
check("等级 5 项且文案是「四级/中级工」这种官方写法",
|
||
jnames == ["五级/初级工", "四级/中级工", "三级/高级工", "二级/技师", "一级/高级技师"],
|
||
f"{jnames}")
|
||
|
||
# ── 3. 驳回原因字典的 itemCode 是 R001 这种码 ──
|
||
print("\n[3] 驳回原因字典：码是 R001，项名是文案")
|
||
rr = items_of(at, "REJECT_REASON")
|
||
codes = [i["itemCode"] for i in rr.get("list") or []]
|
||
check("10 条且码是 R001~R009 + R999",
|
||
codes[:3] == ["R001", "R002", "R003"] and "R999" in codes, f"{codes}")
|
||
check("项名是可读文案（不是码本身）",
|
||
any("材料" in (i.get("itemName") or "") for i in rr.get("list") or []),
|
||
json.dumps((rr.get("list") or [])[:1], ensure_ascii=False))
|
||
|
||
# ── 4. 追加自定义项（对内置字典的追加）──
|
||
print("\n[4] 对内置字典追加自定义项（内置项本身不动）")
|
||
c = body_of("POST", "/api/v1/dicts/EDU_LEVEL/items", at,
|
||
{"itemCode": "E2E_TECH", "itemName": "E2E技师学院"})
|
||
check("追加成功", c.get("code") == 0, json.dumps(c, ensure_ascii=False)[:250])
|
||
new_id = (c.get("data") or {}).get("id")
|
||
check("返回了 id（自定义项有 id，内置项没有）", isinstance(new_id, int), f"{new_id}")
|
||
edu2 = items_of(at, "EDU_LEVEL")
|
||
check("列表里能看到内置 11 项 + 自定义 1 项",
|
||
len(edu2.get("list") or []) == 12, f"{len(edu2.get('list') or [])}")
|
||
check("自定义项标 builtin=false",
|
||
any(i.get("itemCode") == "E2E_TECH" and i.get("builtin") is False
|
||
for i in edu2.get("list") or []))
|
||
ds2 = dict_of(at)
|
||
check("⭐ itemCount 变成 12（含自定义项）",
|
||
(ds2.get("EDU_LEVEL") or {}).get("itemCount") == 12,
|
||
json.dumps(ds2.get("EDU_LEVEL"), ensure_ascii=False))
|
||
|
||
# ── 5. 重复项码要拒 ──
|
||
print("\n[5] 重复项码 / 与内置项撞名")
|
||
dup = body_of("POST", "/api/v1/dicts/EDU_LEVEL/items", at,
|
||
{"itemCode": "E2E_TECH", "itemName": "重复的"})
|
||
check("同字典内项码重复 → 拒绝", dup.get("code") != 0, f"code={dup.get('code')}")
|
||
dup2 = body_of("POST", "/api/v1/dicts/EDU_LEVEL/items", at,
|
||
{"itemCode": "大学本科", "itemName": "想覆盖内置项"})
|
||
check("⭐ 与内置项撞码 → 拒绝，且说明它是内置项",
|
||
dup2.get("code") != 0 and "内置" in (dup2.get("msg") or ""),
|
||
f"code={dup2.get('code')} msg={dup2.get('msg')}")
|
||
bad = body_of("POST", "/api/v1/dicts/EDU_LEVEL/items", at, {"itemCode": "X"})
|
||
check("缺项名 → 拒绝", bad.get("code") != 0, f"code={bad.get('code')}")
|
||
|
||
# ── 6. 改 / 停用 / 恢复 ──
|
||
print("\n[6] 改自定义项 / 停用 / 恢复")
|
||
u = body_of("PUT", f"/api/v1/dicts/items/{new_id}", at,
|
||
{"itemName": "E2E技师学院（改）", "sortNo": 5})
|
||
check("改名成功", u.get("code") == 0, json.dumps(u, ensure_ascii=False)[:200])
|
||
check("库里项名已改",
|
||
sql(f"select item_name from sys_dict_item where id={new_id};").strip()
|
||
== "E2E技师学院（改）")
|
||
d = body_of("PUT", f"/api/v1/dicts/items/{new_id}", at, {"status": 0})
|
||
check("停用成功", d.get("code") == 0, json.dumps(d, ensure_ascii=False)[:200])
|
||
ds3 = dict_of(at)
|
||
check("⭐ 停用后 itemCount 回到 11（只算启用中的项）",
|
||
(ds3.get("EDU_LEVEL") or {}).get("itemCount") == 11,
|
||
json.dumps(ds3.get("EDU_LEVEL"), ensure_ascii=False))
|
||
check("停用后项还在列表里（不是删掉）",
|
||
any(i.get("itemCode") == "E2E_TECH" for i in
|
||
(items_of(at, "EDU_LEVEL").get("list") or [])))
|
||
r = body_of("PUT", f"/api/v1/dicts/items/{new_id}", at, {"status": 1})
|
||
check("重新启用成功", r.get("code") == 0)
|
||
|
||
# ── 7. 内置项只能停用，不能改名 ──
|
||
print("\n[7] 内置项：只能停用，不能改名/删除")
|
||
nb = body_of("PUT", "/api/v1/dicts/items/0", at, {"itemName": "改内置项"})
|
||
check("改不存在的 id → 拒绝", nb.get("code") != 0, f"code={nb.get('code')}")
|
||
# ⚠️ 路径里有中文项码（内置项码就是中文，例如「小学」）→ 必须 URL 编码：
|
||
# http.client 只接受 ASCII 请求行，不编码会直接 UnicodeEncodeError
|
||
enc = urllib.parse.quote("小学")
|
||
t = body_of("POST", f"/api/v1/dicts/EDU_LEVEL/items/{enc}/toggle?enabled=false", at)
|
||
check("停用内置项「小学」成功", t.get("code") == 0, json.dumps(t, ensure_ascii=False)[:250])
|
||
edu3 = items_of(at, "EDU_LEVEL")
|
||
xiao = next((i for i in edu3.get("list") or [] if i.get("itemCode") == "小学"), None)
|
||
check("停用后 status=0（项还在，只是停用）", (xiao or {}).get("status") == 0,
|
||
json.dumps(xiao, ensure_ascii=False))
|
||
ds4 = dict_of(at)
|
||
check("⭐ itemCount 变成 11（内置 10 启用 + 自定义 1）",
|
||
(ds4.get("EDU_LEVEL") or {}).get("itemCount") == 11,
|
||
json.dumps(ds4.get("EDU_LEVEL"), ensure_ascii=False))
|
||
check("停用记在 sys_dict_disabled（不是 sys_dict_item）",
|
||
sql("select count(*) from sys_dict_disabled where dict_code='EDU_LEVEL' "
|
||
"and item_code='小学';").strip() == "1")
|
||
t2 = body_of("POST", f"/api/v1/dicts/EDU_LEVEL/items/{enc}/toggle?enabled=true", at)
|
||
check("⭐ 一键恢复默认可用（这正是把两者分开存的原因）", t2.get("code") == 0,
|
||
json.dumps(t2, ensure_ascii=False)[:200])
|
||
check("恢复后 status 又是 1",
|
||
next((i.get("status") for i in (items_of(at, "EDU_LEVEL").get("list") or [])
|
||
if i.get("itemCode") == "小学"), None) == 1)
|
||
bad2 = body_of("POST", "/api/v1/dicts/EDU_LEVEL/items/E2E_TECH/toggle?enabled=false", at)
|
||
check("对**自定义**项调内置 toggle → 拒绝（说清它不是内置项）",
|
||
bad2.get("code") != 0 and "内置" in (bad2.get("msg") or ""),
|
||
f"code={bad2.get('code')} msg={bad2.get('msg')}")
|
||
|
||
# ── 8. 层级：有下级项的项不可停用（40036）──
|
||
print("\n[8] 层级（40036：有下级项的项不可停用）")
|
||
p = body_of("POST", f"/api/v1/dicts/{TEST_DICT}/items", at,
|
||
{"itemCode": "P1", "itemName": "父项"})
|
||
check("建父项成功（自定义字典码会被隐式创建）", p.get("code") == 0,
|
||
json.dumps(p, ensure_ascii=False)[:200])
|
||
pid = (p.get("data") or {}).get("id")
|
||
ch = body_of("POST", f"/api/v1/dicts/{TEST_DICT}/items", at,
|
||
{"itemCode": "C1", "itemName": "子项", "parentItemCode": "P1"})
|
||
check("建子项成功", ch.get("code") == 0, json.dumps(ch, ensure_ascii=False)[:200])
|
||
cid = (ch.get("data") or {}).get("id")
|
||
check("自定义字典出现在字典列表里且 builtin=false",
|
||
(dict_of(at).get(TEST_DICT) or {}).get("builtin") is False,
|
||
json.dumps(dict_of(at).get(TEST_DICT), ensure_ascii=False))
|
||
items = items_of(at, TEST_DICT)
|
||
check("字典项接口标出这个字典支持层级", items.get("hierarchical") is True,
|
||
json.dumps(items, ensure_ascii=False)[:200])
|
||
stop = body_of("PUT", f"/api/v1/dicts/items/{pid}", at, {"status": 0})
|
||
check("⭐ 停用父项 → 40036（有下级不可停用）", stop.get("code") == 40036,
|
||
f"code={stop.get('code')} msg={stop.get('msg')}")
|
||
dele = body_of("DELETE", f"/api/v1/dicts/items/{pid}", at)
|
||
check("⭐ 删父项 → 同样被拦（40036）", dele.get("code") == 40036,
|
||
f"code={dele.get('code')} msg={dele.get('msg')}")
|
||
badp = body_of("POST", f"/api/v1/dicts/{TEST_DICT}/items", at,
|
||
{"itemCode": "C2", "itemName": "孤儿", "parentItemCode": "NOT_EXIST"})
|
||
check("父项不存在 → 拒绝（不造孤儿项）", badp.get("code") != 0,
|
||
f"code={badp.get('code')} msg={badp.get('msg')}")
|
||
dl = body_of("DELETE", f"/api/v1/dicts/items/{cid}", at)
|
||
check("先删子项成功", dl.get("code") == 0, json.dumps(dl, ensure_ascii=False)[:200])
|
||
dl2 = body_of("DELETE", f"/api/v1/dicts/items/{pid}", at)
|
||
check("再删父项成功（下级没了就能删）", dl2.get("code") == 0)
|
||
check("软删（行还在，deleted=1）",
|
||
sql(f"select count(*) from sys_dict_item where id={pid} and deleted=1;").strip() == "1")
|
||
check("⭐ 删掉后同项码能重新加（部分唯一索引只约束未删的行）",
|
||
body_of("POST", f"/api/v1/dicts/{TEST_DICT}/items", at,
|
||
{"itemCode": "P1", "itemName": "父项重建"}).get("code") == 0)
|
||
|
||
# ── 9. 项码不可改 ──
|
||
print("\n[9] 项码不可改（改了会破坏历史数据）")
|
||
check("修改接口的 DTO 里根本没有 itemCode 字段（结构性保证）",
|
||
"itemCode" not in json.dumps(
|
||
body_of("PUT", f"/api/v1/dicts/items/{new_id}", at,
|
||
{"itemName": "E2E技师学院（改）"}).get("data") or {}))
|
||
body_of("PUT", f"/api/v1/dicts/items/{new_id}", at, {"itemCode": "HACKED"})
|
||
check("⭐ 传了 itemCode 也不生效（库里项码没变）",
|
||
sql(f"select item_code from sys_dict_item where id={new_id};").strip() == "E2E_TECH",
|
||
sql(f"select item_code from sys_dict_item where id={new_id};"))
|
||
|
||
# ── 10. 权限 ──
|
||
print("\n[10] 权限")
|
||
cfg = R.read_dev_config()
|
||
stok = R.mint_mp_token(cfg["jwt_secret"], cfg["issuer"], 993901, 1, ["STUDENT"], 4)
|
||
b = body_of("POST", "/api/v1/dicts/EDU_LEVEL/items", stok,
|
||
{"itemCode": "HACK", "itemName": "学生想加字典"})
|
||
check("学生加字典项 → 403", b.get("code") in (40101, 40301, 403),
|
||
f"code={b.get('code')} msg={b.get('msg')}")
|
||
b2 = body_of("GET", "/api/v1/dicts", stok)
|
||
check("学生读字典 → 也拒（字典是配置面，不是业务面）",
|
||
b2.get("code") in (40101, 40301, 403),
|
||
f"code={b2.get('code')} msg={b2.get('msg')}")
|
||
b3 = body_of("GET", "/api/v1/dicts")
|
||
check("未登录 → 40101", b3.get("code") == 40101, f"code={b3.get('code')}")
|
||
|
||
cleanup()
|
||
print("\n" + "=" * 74)
|
||
print(f"通过 {len(PASSED)} 失败 {len(FAILED)} 共 {len(PASSED) + len(FAILED)}")
|
||
print("=" * 74)
|
||
for n, det in FAILED:
|
||
print(f" ✗ {n}\n {det}")
|
||
return 1 if FAILED else 0
|
||
|
||
|
||
def cleanup() -> None:
|
||
R.sql(f"delete from sys_dict_item where tenant_id={TENANT} "
|
||
f"and (dict_code='{TEST_DICT}' or item_code like 'E2E%');")
|
||
R.sql(f"delete from sys_dict_disabled where tenant_id={TENANT};")
|
||
print(" 已清理：测试字典项 / 内置项停用记录")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|