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