# -*- 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 unittest ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) 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: ddl = fh.read() 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) class TestCrudJsonContract(unittest.TestCase): """json/pbl_domain_ref.json 与 models 字段名一致（QC#1 同步项）。""" @classmethod def setUpClass(cls): cls.crud = _load(CRUD_PATH) cls.model_fields = {f["name"] for f in _load(MODEL_PATH)["fields"]} def test_tblname_and_editable(self): self.assertEqual(self.crud["tblname"], "pbl_domain_ref") params = self.crud["params"] for key in ("new_data_url", "update_data_url", "delete_data_url"): self.assertIn(key, params, "params 顶层缺 %s" % key) self.assertIn("entire_url", params[key]) self.assertIn("editable", params) def test_browserfields_subset_of_model(self): bf = set(self.crud["params"]["browserfields"].keys()) self.assertEqual(bf - self.model_fields, set(), "CRUD 出现 models 未声明的字段: %s" % (bf - self.model_fields)) def test_editexclouded_known_columns(self): for col in self.crud["params"].get("editexclouded", []): self.assertIn(col, self.model_fields) def test_no_new_data_url_query_params(self): """new_data_url 不得带 query 参数（避免与 editexclouded 合并成 list）。""" url = self.crud["params"]["new_data_url"] self.assertNotIn("?", url) 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() self.assertEqual(len(a), 32) 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 不应被调用") import asyncio from pbl_domain_ext import db as dbm old = dbm._SOR_HOLDER.get("sor") dbm.set_sor(FakeSor()) try: asyncio.get_event_loop().run_until_complete( 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(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)