# -*- coding: utf-8 -*- """机械契约测试：models/pbl_domain_ref.json ↔ pbl_domain_ext 代码列名/类型一致性。 覆盖 QC 退回意见 #1/#2/#5/#6 的回归防护： #1 代码读写的 7 个列（ref_code/ref_name/bind_state/bind_at/is_deleted/creator_id/updater_id） 必须全部出现在 models fields 中，且审计列名不得写成 created_by/updated_by； #2 ref_id/blueprint_id 必须是 str 类型（与基表 world/scene/entity 主键 str(32) 一致）， 禁止 long/bigint； #5 init/data.json 必须是合法 JSON，且按 Format B 注入 appcodes 两组 （pbl_domain_ref_type / pbl_bind_state），parentid 与 models codes cond 对得上； #6 主键生成必须走 appPublic.uniqueID.getID（禁止 uuid4）。 """ import json import os import re import sys import unittest ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if ROOT not in sys.path: sys.path.insert(0, ROOT) # 允许 python3 tests/test_models_contract.py 直接跑 MODEL_PATH = os.path.join(ROOT, "models", "pbl_domain_ref.json") CRUD_PATH = os.path.join(ROOT, "json", "pbl_domain_ref.json") INIT_DATA_PATH = os.path.join(ROOT, "init", "data.json") SQL_PATH = os.path.join(ROOT, "sql", "pbl_domain_ext.sql") #: api.py / db.py / base.py 实际读写的列（QC#1 清单） CODE_USED_COLUMNS = [ "id", "tenant_id", "ref_type", "ref_id", "ref_code", "ref_name", "blueprint_id", "class_id", "team_id", "bind_state", "bind_at", "ext_json", "is_deleted", "creator_id", "created_at", "updater_id", "updated_at", ] FORBIDDEN_AUDIT_NAMES = ["created_by", "updated_by", "creator", "updater", "ext"] def _load(path): with open(path, "r", encoding="utf-8") as fh: return json.load(fh) class TestModelContract(unittest.TestCase): """models/pbl_domain_ref.json 与代码契约同名同型。""" @classmethod def setUpClass(cls): cls.model = _load(MODEL_PATH) cls.fields = {f["name"]: f for f in cls.model["fields"]} def test_summary_shape(self): """summary 必须恰一条、primary 必须是数组 ["id"]（table-definition-spec）。""" self.assertEqual(len(self.model["summary"]), 1) s = self.model["summary"][0] self.assertEqual(s["name"], "pbl_domain_ref") self.assertIsInstance(s["primary"], list) self.assertEqual(s["primary"], ["id"]) def test_only_one_table(self): """Q-OPEN-3 薄扩展：models/ 下只允许 pbl_domain_ref 一张表， 禁止把 world/scene/entity 基表重新定义进来。""" names = sorted(f for f in os.listdir(os.path.join(ROOT, "models")) if f.endswith(".json")) self.assertEqual(names, ["pbl_domain_ref.json"]) def test_code_columns_all_declared(self): """QC#1：代码读写的列必须全部在 fields 里声明。""" missing = [c for c in CODE_USED_COLUMNS if c not in self.fields] self.assertEqual(missing, [], "models 缺列: %s" % missing) def test_no_forbidden_column_names(self): """QC#1：审计列名统一 creator_id/updater_id，扩展列统一 ext_json。""" bad = [n for n in self.fields if n in FORBIDDEN_AUDIT_NAMES] self.assertEqual(bad, [], "存在旧名/禁用列名: %s" % bad) def test_ref_id_and_blueprint_id_are_str(self): """QC#2：ref_id / blueprint_id 必须 str，长度 >= 32，禁止 long/bigint。""" for col, minlen in (("ref_id", 32), ("blueprint_id", 32), ("id", 32)): f = self.fields[col] self.assertEqual(f["type"], "str", "%s 类型应为 str" % col) self.assertIsInstance(f["length"], int) self.assertGreaterEqual(f["length"], minlen) for col in self.fields: self.assertNotIn(self.fields[col]["type"], ("long", "bigint"), "%s 不得使用 bigint" % col) def test_str_fields_have_int_length(self): """str/char 必须带正整数 length（禁止 "15,2" 之类字符串写法）。""" for name, f in self.fields.items(): if f["type"] in ("str", "char"): self.assertIsInstance(f.get("length"), int, "%s 缺 int length" % name) self.assertGreater(f["length"], 0) if f["type"] in ("float", "double", "ddouble", "decimal"): self.assertIsInstance(f.get("length"), int) self.assertIsInstance(f.get("dec"), int) def test_indexes_shape(self): """索引必须用 idxfields 数组；唯一键 = (tenant_id, ref_type, ref_id)。""" names = set() for idx in self.model["indexes"]: self.assertIn("idxtype", idx) self.assertIsInstance(idx["idxfields"], list) self.assertNotIn(idx["name"], names) names.add(idx["name"]) uniq = [i for i in self.model["indexes"] if i["idxtype"] == "unique"] self.assertEqual(len(uniq), 1) self.assertEqual(uniq[0]["idxfields"], ["tenant_id", "ref_type", "ref_id"]) def test_codes_use_parentid(self): """codes 引用 appcodes_kv 必须 cond parentid=，禁止 id=；禁止 module.table 点号。""" for c in self.model["codes"]: self.assertNotIn(".", c["table"]) if c["table"] == "appcodes_kv": self.assertTrue(c["cond"].startswith("parentid="), "codes cond 必须 parentid= : %s" % c) fields = [c["field"] for c in self.model["codes"]] self.assertEqual(len(fields), len(set(fields)), "codes 存在重复 field") def test_list_fields_match_model(self): """base.LIST_FIELDS 必须与 models fields 完全一致（无多无少）。""" from pbl_domain_ext.base import LIST_FIELDS, ALL_COLUMNS, AUDIT_FIELDS self.assertEqual(sorted(LIST_FIELDS), sorted(self.fields.keys())) self.assertEqual(sorted(ALL_COLUMNS), sorted(self.fields.keys())) for a in AUDIT_FIELDS: self.assertIn(a, self.fields) def test_sql_ddl_columns_match_model(self): """sql/pbl_domain_ext.sql 的建表列必须覆盖 models 全部列。""" if not os.path.exists(SQL_PATH): self.skipTest("sql/pbl_domain_ext.sql 不存在（由 json2ddl 生成）") with open(SQL_PATH, "r", encoding="utf-8") as fh: raw_ddl = fh.read() # 剥离 -- 行注释与 /* */ 块注释：迁移说明注释里合法提到旧列名， # 只有**生效语句**里的列名才是契约。 ddl = re.sub(r"/\*.*?\*/", "", raw_ddl, flags=re.S) ddl = "\n".join(ln for ln in ddl.splitlines() if not ln.strip().startswith("--")) for col in self.fields: self.assertRegex(ddl, r"`%s`" % col, "DDL 缺列 %s" % col) for bad in FORBIDDEN_AUDIT_NAMES: self.assertNotRegex(ddl, r"`%s`" % bad, "DDL 含禁用旧列名 %s" % bad) # 生效建表语句必须恰为 1 张表（Q-OPEN-3 薄扩展：只新增 pbl_domain_ref） created = re.findall(r"CREATE TABLE(?: IF NOT EXISTS)?\s+`?(\w+)`?", ddl, re.I) self.assertEqual(created, ["pbl_domain_ref"], "薄扩展只允许建 1 张 pbl_domain_ref，实得 %s" % created) class TestCrudJsonContract(unittest.TestCase): """json/pbl_domain_ref.json 符合 crud-definition-spec（QC#4）并与 models 字段名一致。""" #: crud-definition-spec 里 params 不允许出现的自创键 FORBIDDEN_PARAM_KEYS = ("filters", "sortname", "sortorder", "rows", "fields", "columns", "grid", "form", "components", "select_fields") @classmethod def setUpClass(cls): cls.crud = _load(CRUD_PATH) cls.model_fields = {f["name"] for f in _load(MODEL_PATH)["fields"]} cls.params = cls.crud["params"] def test_tblname_and_editable(self): self.assertEqual(self.crud["tblname"], "pbl_domain_ref") for key in ("new_data_url", "update_data_url", "delete_data_url"): self.assertIn(key, self.params, "params 顶层缺 %s" % key) self.assertIn("entire_url", self.params[key]) self.assertIn("editable", self.params) def test_no_forbidden_root_keys(self): """根键只允许 tblname/alias/title/params；禁止 tablename/grid/form/name/type。""" self.assertNotIn("tablename", self.crud) for bad in ("tablename", "grid", "form", "name", "type", "components"): self.assertNotIn(bad, self.crud) self.assertNotIn(bad, self.params) def test_no_self_invented_param_keys(self): """QC#4：filters/sortname/sortorder/rows 是自创键，规范用 sortby + data_filter。""" for bad in self.FORBIDDEN_PARAM_KEYS: self.assertNotIn(bad, self.params, "params 含自创键 %s" % bad) self.assertIsInstance(self.params.get("sortby"), list, "排序必须用 sortby 数组") self.assertIn("data_filter", self.params, "筛选必须用 data_filter") def test_data_filter_shape(self): """data_filter 必须是 {AND:[{field,op,var|const}]}，且 field 都在 models 中。""" df = self.params["data_filter"] self.assertIn("AND", df) self.assertGreaterEqual(len(df["AND"]), 2, "AND 数组长度须 >= 2") for cond in df["AND"]: self.assertIn(cond["field"], self.model_fields, "data_filter 引用未声明列 %s" % cond["field"]) self.assertIn(cond["op"], ("=", "!=", ">", ">=", "<", "<=", "IN", "NOT IN", "LIKE", "NOT LIKE", "IS NULL", "IS NOT NULL")) self.assertTrue("var" in cond or "const" in cond, "条件必须带 var 或 const: %s" % cond) def test_browserfields_structure(self): """QC#4：browserfields 结构为 {exclouded:[...], alters:{field:{uitype,...}}}， 不是平铺字段定义。""" bf = self.params["browserfields"] self.assertIsInstance(bf, dict) self.assertIn("alters", bf, "browserfields 缺 alters 段（平铺写法不符合规范）") self.assertIn("exclouded", bf, "browserfields 缺 exclouded 段") self.assertIsInstance(bf["exclouded"], list) for col in bf["exclouded"]: self.assertIn(col, self.model_fields, "exclouded 列 %s 未在 models 声明" % col) for col, spec in bf["alters"].items(): self.assertIn(col, self.model_fields, "alters 列 %s 未在 models 声明" % col) self.assertIsInstance(spec, dict) self.assertIn("uitype", spec, "alters.%s 缺 uitype" % col) if spec["uitype"] == "code": self.assertTrue(spec.get("dataurl") or spec.get("data"), "code 型字段 %s 必须给 dataurl 或 data" % col) if spec.get("dataurl"): self.assertIn("entire_url", spec["dataurl"]) def test_exclouded_not_at_params_top(self): """QC#4：exclouded 属于 browserfields，不得放在 params 顶层。""" self.assertNotIn("exclouded", self.params) def test_editexclouded_known_columns(self): for col in self.params.get("editexclouded", []): self.assertIn(col, self.model_fields, "editexclouded 列 %s 未在 models 声明" % col) def test_editexclouded_covers_not_null_defaults(self): """NOT NULL 且非用户可编辑的列必须进 editexclouded，否则提交报 cannot be null。""" model = _load(MODEL_PATH) auto = {"id", "tenant_id", "creator_id", "created_at", "updater_id", "updated_at", "is_deleted", "bind_at", "ref_type", "ref_id", "ref_code", "ref_name"} edit = set(self.params.get("editexclouded", [])) for f in model["fields"]: if f.get("nullable") == "no" and f["name"] in auto: self.assertIn(f["name"], edit, "NOT NULL 列 %s 未 editexclouded" % f["name"]) def test_data_url_not_duplicated_in_editable(self): """QC#12：三个写 URL 只在 params 顶层一份，禁止在 editable 内重复定义。 重复定义会让 xls2crud 模板读取歧义（历史上 discount 模块因嵌套导致 自定义 create 逻辑被绕过）。editable 只允许保留 get_data_url。 """ editable = self.params.get("editable", {}) for key in ("new_data_url", "update_data_url", "delete_data_url"): self.assertNotIn(key, editable, "params.editable 内不得重复定义 %s（权威位置=params 顶层）" % key) self.assertIn("get_data_url", editable, "editable 缺 get_data_url（列表取数地址）") def test_field_maxlen_matches_model(self): """QC#23：base.FIELD_MAXLEN 必须与 models/*.json 的 length 单一来源一致。""" import sys sys.path.insert(0, os.path.join(ROOT, "pbl_domain_ext")) from pbl_domain_ext import base for f in _load(MODEL_PATH)["fields"]: name = f["name"] self.assertIn(name, base.FIELD_MAXLEN, "FIELD_MAXLEN 缺列 %s" % name) if f.get("type") in ("str", "char") and f.get("length"): self.assertEqual(base.FIELD_MAXLEN[name], f["length"], "FIELD_MAXLEN[%s]=%s != model length=%s" % (name, base.FIELD_MAXLEN[name], f["length"])) else: self.assertIsNone(base.FIELD_MAXLEN[name], "非字符列 %s 不应有截断宽度" % name) def test_clip_respects_column_width(self): """QC#23：clip() 按列宽裁剪——blueprint_id=32、ref_code=64、ref_name=128。""" from pbl_domain_ext import base self.assertEqual(len(base.clip("blueprint_id", "b" * 80)), 32) self.assertEqual(len(base.clip("ref_code", "c" * 80)), 64) self.assertEqual(len(base.clip("class_id", "x" * 80)), 64) self.assertEqual(len(base.clip("ref_name", "n" * 200)), 128) self.assertEqual(base.clip("ref_code", " w1 "), "w1") # 非字符列（text/timestamp）不截断 self.assertEqual(base.clip("ext_json", "e" * 5000)[:5], "eeeee") def test_no_new_data_url_query_params(self): """new_data_url 不得带 query 参数（避免与 editexclouded 合并成 list）。""" url = self.params["new_data_url"] self.assertNotIn("?", url) def test_no_jinja_control_blocks(self): """CRUD JSON 禁止 Jinja2 控制块，只允许 {{entire_url(...)}} 插值。""" with open(CRUD_PATH, "r", encoding="utf-8") as fh: raw = fh.read() self.assertIsNone(re.search(r"\{%\s*(if|for|each|end)", raw), "CRUD JSON 含 Jinja2 控制块") def test_editable_urls_have_dspy_endpoints(self): """每个 editable/data_url 指向的 .dspy 必须真实存在于 wwwroot/api/。""" names = [] for key in ("new_data_url", "update_data_url", "delete_data_url"): names.append(self.params[key]) for spec in self.params["browserfields"]["alters"].values(): du = spec.get("dataurl", "") if "pbl_domain_ext" in du: names.append(du) api_dir = os.path.join(ROOT, "wwwroot", "api") existing = set(os.listdir(api_dir)) for url in names: m = re.search(r"entire_url\('([^']+)'\)", url) self.assertIsNotNone(m, "URL 未用 entire_url 包裹: %s" % url) fn = m.group(1).rstrip("/").split("/")[-1] if "." not in fn: # CRUD alias 目录，由 xls2ui 生成 continue self.assertIn(fn, existing, "缺少 .dspy 端点文件: %s" % fn) class TestInitDataContract(unittest.TestCase): """QC#5：init/data.json Format B 种子必须真实存在且与 codes cond 对齐。""" @classmethod def setUpClass(cls): cls.data = _load(INIT_DATA_PATH) cls.codes = _load(MODEL_PATH)["codes"] def test_is_valid_json_format_b(self): self.assertIn("appcodes", self.data) self.assertIsInstance(self.data["appcodes"], list) self.assertGreaterEqual(len(self.data["appcodes"]), 2) def test_parentids_cover_codes_cond(self): parents = {g["parentid"] for g in self.data["appcodes"]} for c in self.codes: if c["table"] != "appcodes_kv": continue m = re.search(r"parentid='([^']+)'", c["cond"]) self.assertIsNotNone(m, "codes cond 解析失败: %s" % c) self.assertIn(m.group(1), parents, "codes 引用的编码组 %s 未在 init/data.json 注入" % m.group(1)) def test_items_and_id_length(self): """每组必须有 items(k/v)；parentid+k 生成的 id 不得超 VARCHAR(32)。""" for g in self.data["appcodes"]: self.assertLessEqual(len(g["parentid"]), 22, "parentid 过长会导致 appcodes_kv.id 超 32: %s" % g["parentid"]) self.assertTrue(g.get("items"), "%s 无 items" % g["parentid"]) for it in g["items"]: self.assertTrue(it.get("k") and it.get("v")) self.assertLessEqual(len("%s_%s" % (g["parentid"], it["k"])), 32) def test_ref_type_items_match_code_enum(self): """world/scene/entity 三值必须齐（与 base.REF_TYPES 一致）。""" from pbl_domain_ext.base import REF_TYPES, BIND_STATES groups = {g["parentid"]: {it["k"] for it in g["items"]} for g in self.data["appcodes"]} self.assertEqual(groups["pbl_domain_ref_type"], set(REF_TYPES)) self.assertEqual(groups["pbl_bind_state"], set(BIND_STATES)) class TestIdGeneration(unittest.TestCase): """QC#6：主键必须走 appPublic.uniqueID.getID，禁止 uuid4。""" def test_no_uuid_in_package(self): pkg = os.path.join(ROOT, "pbl_domain_ext") for fn in sorted(os.listdir(pkg)): if not fn.endswith(".py"): continue with open(os.path.join(pkg, fn), "r", encoding="utf-8") as fh: src = fh.read() self.assertNotIn("uuid", src, "%s 不得使用 uuid 生成主键" % fn) def test_base_uses_platform_getid(self): with open(os.path.join(ROOT, "pbl_domain_ext", "base.py"), "r", encoding="utf-8") as fh: src = fh.read() self.assertIn("from appPublic.uniqueID import getID", src) def test_gen_id_shape(self): from pbl_domain_ext.base import gen_id a, b = gen_id(), gen_id() # 平台 appPublic.uniqueID.getID() 实际返回 21 位；离线降级路径 32 位 hex。 # 两者都在 models 列宽 str(32) 内，断言按「唯一 + 不超列宽」而非固定长度。 self.assertGreaterEqual(len(a), 16, "主键过短，碰撞风险不可接受: %r" % a) self.assertLessEqual(len(a), 32, "主键超列宽 str(32): %r" % a) self.assertNotEqual(a, b) class TestHostAgnosticImports(unittest.TestCase): """QC#4：模块宿主无关——禁止 import 具体宿主应用（sage/pipeline_app 等）。""" FORBIDDEN_HOSTS = ("sage", "pipeline_app", "hrs6", "hrs7") ALLOWED_ROOTS = ("ahserver", "sqlor", "appPublic", "apppublic", "pbl_domain_ext", "json", "os", "re", "sys", "time", "hashlib", "unittest") def test_no_host_import(self): pkg = os.path.join(ROOT, "pbl_domain_ext") pat = re.compile(r"^\s*(?:from|import)\s+([A-Za-z_][\w\.]*)", re.M) for fn in sorted(os.listdir(pkg)): if not fn.endswith(".py"): continue with open(os.path.join(pkg, fn), "r", encoding="utf-8") as fh: src = fh.read() for mod in pat.findall(src): top = mod.split(".")[0] self.assertNotIn(top, self.FORBIDDEN_HOSTS, "%s 违反宿主无关铁律: import %s" % (fn, mod)) def test_serverenv_from_ahserver(self): with open(os.path.join(ROOT, "pbl_domain_ext", "db.py"), "r", encoding="utf-8") as fh: src = fh.read() self.assertIn("from ahserver.serverenv import ServerEnv", src) self.assertNotIn("from sage", src) class TestSqlorSignatures(unittest.TestCase): """QC#3：sor.C/U/D/R 只收 2 参——真实签名断言（不再被 fake_db 掩盖）。""" def test_no_three_arg_crud_calls(self): pkg = os.path.join(ROOT, "pbl_domain_ext") bad = re.compile(r"sor\.[CUDR]\(\s*[^()]*?,\s*[^()]*?,\s*[^()]*?\)") for fn in sorted(os.listdir(pkg)): if not fn.endswith(".py"): continue with open(os.path.join(pkg, fn), "r", encoding="utf-8") as fh: src = fh.read() self.assertIsNone(bad.search(src), "%s 存在 3 参 sor.C/U/D/R 调用" % fn) def test_sor_I_single_arg(self): pkg = os.path.join(ROOT, "pbl_domain_ext") bad = re.compile(r"sor\.I\(\s*[^()]*?,") for fn in sorted(os.listdir(pkg)): if not fn.endswith(".py"): continue with open(os.path.join(pkg, fn), "r", encoding="utf-8") as fh: src = fh.read() self.assertIsNone(bad.search(src), "%s sor.I 只能 1 参" % fn) def test_update_ref_merges_where_into_ns(self): """update_ref 必须把 where 并入 ns 后两参调用 sor.U。""" calls = [] class FakeSor(object): async def U(self, table, ns): calls.append((table, dict(ns))) return 1 def C(self, *a, **k): raise AssertionError("C 不应被调用") from pbl_domain_ext import db as dbm old = dbm._SOR_HOLDER.get("sor") dbm.set_sor(FakeSor()) try: # _sor_u 是同步契约函数：内部 _drive 负责把 async sor.U 的协程收敛掉， # 测试不得再用 run_until_complete 包一层（协程已被 _drive 消费， # run_until_complete(None) 会抛 TypeError —— 旧断言的错误写法）。 rc = dbm._sor_u(FakeSor(), "pbl_domain_ref", {"bind_state": "unbound", "updater_id": "u1"}, {"id": "pk1", "tenant_id": "t1"}) finally: dbm._SOR_HOLDER["sor"] = old self.assertEqual(int(rc or 0), 1, "应返回受影响行数") self.assertEqual(len(calls), 1) table, ns = calls[0] self.assertEqual(table, "pbl_domain_ref") # where 条件必须在 ns 内，且不被 SET 覆盖 self.assertEqual(ns["id"], "pk1") self.assertEqual(ns["tenant_id"], "t1") self.assertEqual(ns["bind_state"], "unbound") if __name__ == "__main__": unittest.main(verbosity=2)