diff --git a/.gitignore b/.gitignore index 6579d96..05d9152 100644 --- a/.gitignore +++ b/.gitignore @@ -10,3 +10,8 @@ dist/ build/ .venv/ venv/ +_patch*.py +ut.txt +*.orig +*.rej +tests/__pycache__/ diff --git a/_patch2.py b/_patch2.py deleted file mode 100644 index 024fbfe..0000000 --- a/_patch2.py +++ /dev/null @@ -1,179 +0,0 @@ -# -*- coding: utf-8 -*- -"""一次性修补：fake_db 真实 2 参签名 / api 契约修正 / 陈旧测试列名对齐。""" -import io -import os - -ROOT = os.path.dirname(os.path.abspath(__file__)) - - -def patch(path, pairs, required=True): - p = os.path.join(ROOT, path) - src = io.open(p, encoding="utf-8").read() - for old, new in pairs: - if old not in src: - if required: - raise SystemExit("NOT FOUND in %s:\n%s" % (path, old[:200])) - continue - src = src.replace(old, new, 1) - io.open(p, "w", encoding="utf-8").write(src) - print("patched", path) - - -# ---------------------------------------------------------------- fake_db -# QC#3：fake_db 的 U 曾是 3 参 (tbl,row,where)，掩盖了真实 sqlor 2 参签名。 -# 现改为真实签名 U(tbl, ns)：ns 同时含 SET 值与 WHERE 条件（条件键 = id/tenant_id）。 -patch("tests/fake_db.py", [ - (""" def U(self, tbl, row, where): - self._guard_readonly(tbl, "U") - hits = 0 - for exist in self.tables.get(tbl, []): - if self._match(exist, where): - exist.update(row) - hits += 1 - self.sql_log.append(("U", tbl, dict(row), dict(where))) - return hits""", - """ #: 真实 sqlor U(table, ns) 只收 2 参——ns 同时携带 SET 值与 WHERE 条件。 - #: 条件键约定为主键/租户键（本模块 update_ref 的 row 从不含这两列）。 - WHERE_KEYS = ("id", "tenant_id") - - def U(self, tbl, ns): - \"\"\"真实 sqlor 签名：U(table, ns)，2 参（crud-spec Pitfall 22）。\"\"\" - self._guard_readonly(tbl, "U") - ns = dict(ns or {}) - where = {k: ns.pop(k) for k in self.WHERE_KEYS if k in ns} - if not where: - raise AssertionError( - "sqlor U(table, ns) 的 ns 必须含 WHERE 条件键 %s，实得 %s" - % (list(self.WHERE_KEYS), sorted(ns))) - hits = 0 - for exist in self.tables.get(tbl, []): - if self._match(exist, where): - exist.update(ns) - hits += 1 - self.sql_log.append(("U", tbl, dict(ns), where)) - return hits"""), -]) - -# ---------------------------------------------------------------- api.py -# ① world_get_context：基表未命中/跨租户必须报错，不得静默返回空上下文 -patch("pbl_domain_ext/api.py", [ - (""" base = db.read_base_row("world", world_id, tenant_id) - world_view = _base_view(base or {}) - ref = db.select_ref_by_key(tenant_id, "world", world_id)""", - """ base = db.read_base_row("world", world_id, tenant_id) - world_view = _base_view(base or {}) - if not world_view: - # 基表未命中或 world 属其它租户：一律按不存在处理（跨租户不得静默成功） - raise PblDomainExtError( - "PBL_DE_BASE_MISSING", - detail="world.%s 在 tenant_id=%s 下未命中" % (world_id, tenant_id)) - ref = db.select_ref_by_key(tenant_id, "world", world_id)"""), - # ② team_world_list：本端点语义是「列出世界」，item.id = world id； - # 关联表主键另置 ref_pk，避免调用方把关联主键当世界 ID。 - (""" for r in rows: - item = _ref_out(r) - item["world_id"] = item.get("ref_id") - item["world_code"] = item.get("ref_code") - item["world_name"] = item.get("ref_name") - items.append(item)""", - """ for r in rows: - item = _ref_out(r) - item["ref_pk"] = item.get("id") # 关联表主键 - item["id"] = item.get("ref_id") # 对外身份 = world.id - item["world_id"] = item.get("ref_id") - item["world_code"] = item.get("ref_code") - item["world_name"] = item.get("ref_name") - items.append(item)"""), -]) - -# ---------------------------------------------------------------- 陈旧测试列名 -# QC#1 定案：审计列权威名 creator_id/updater_id（models 与代码同名同型）。 -patch("tests/test_domain_ref.py", [ - ('self.assertEqual(data["created_by"], OP)', - 'self.assertEqual(data["creator_id"], OP) # QC#1 权威列名 creator_id'), - ('self.assertEqual(self.rows()[0]["updated_by"], OP)', - 'self.assertEqual(self.rows()[0]["updater_id"], OP) # QC#1 权威列名 updater_id'), -]) - -# ---------------------------------------------------------------- 契约测试自身 -patch("tests/test_models_contract.py", [ - (""" 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")""", - """ calls = [] - - class AsyncSor(object): - \"\"\"真实 sqlor 形态：C/U/D/R 均为 async 且只收 2 参。\"\"\" - - async def U(self, table, ns): - calls.append((table, dict(ns))) - return 1 - - async def C(self, table, ns): - raise AssertionError("C 不应被调用") - - class SyncSor(AsyncSor): - \"\"\"同步形态（离线 fake_db）——_drive 必须同样收敛。\"\"\" - - def U(self, table, ns): - calls.append((table, dict(ns))) - return 1 - - from pbl_domain_ext import db as dbm - for sor in (AsyncSor(), SyncSor()): - del calls[:] - affected = dbm._sor_u(sor, "pbl_domain_ref", - {"bind_state": "unbound", "updater_id": "u1"}, - {"id": "pk1", "tenant_id": "t1"}) - self.assertEqual(affected, 1, "返回值必须收敛为普通值，不得是 coroutine") - 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") - - def test_drive_converges_awaitable(self): - \"\"\"QC#3 回归：宿主 async sqlor 下不得把 coroutine 泄漏给 int()/调用方。\"\"\" - from pbl_domain_ext.db import _drive - - async def _coro(): - return 7 - - self.assertEqual(_drive(_coro()), 7) - self.assertEqual(_drive(3), 3) - - def test_update_ref_rejects_empty_where(self): - from pbl_domain_ext import db as dbm - from pbl_domain_ext.errors import PblDomainExtError - with self.assertRaises(PblDomainExtError): - dbm.update_ref({}, {"bind_state": "unbound"})"""), -]) - -print("all patches applied") diff --git a/json/pbl_domain_ref.json b/json/pbl_domain_ref.json index 4b40731..808ac3f 100644 --- a/json/pbl_domain_ref.json +++ b/json/pbl_domain_ref.json @@ -10,51 +10,64 @@ "update_data_url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_update.dspy')}}", "delete_data_url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_unbind.dspy')}}" }, + "sortby": ["bind_at desc", "id desc"], + "data_filter": { + "AND": [ + {"field": "tenant_id", "op": "=", "var": "tenant_id_input"}, + {"field": "ref_type", "op": "=", "var": "ref_type_input"}, + {"field": "bind_state", "op": "=", "var": "bind_state_input"}, + {"field": "blueprint_id", "op": "=", "var": "blueprint_id_input"}, + {"field": "class_id", "op": "=", "var": "class_id_input"}, + {"field": "team_id", "op": "=", "var": "team_id_input"} + ] + }, + "filter_labels": { + "tenant_id_input": "租户ID", + "ref_type_input": "关联对象类型", + "bind_state_input": "绑定状态", + "blueprint_id_input": "关联蓝图ID", + "class_id_input": "班级ID", + "team_id_input": "团队ID" + }, + "filter_label": "搜索", + "filter_title": "关联记录筛选", + "confidential_fields": ["ext_json"], "browserfields": { - "id": {"title": "主键", "uitype": "text", "width": 120}, - "tenant_id": {"title": "租户ID", "uitype": "text", "width": 120}, - "ref_type": { - "title": "关联对象类型", - "uitype": "code", - "width": 110, - "dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_domain_ref_type%27')}}" - }, - "ref_id": {"title": "基表记录ID", "uitype": "text", "width": 160}, - "ref_code": {"title": "基表编码快照", "uitype": "text", "width": 130}, - "ref_name": {"title": "基表名称快照", "uitype": "text", "width": 180}, - "blueprint_id": {"title": "关联蓝图ID", "uitype": "text", "width": 150}, - "class_id": {"title": "班级ID", "uitype": "text", "width": 120}, - "team_id": {"title": "团队ID", "uitype": "text", "width": 120}, - "bind_state": { - "title": "绑定状态", - "uitype": "code", - "width": 90, - "dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_bind_state%27')}}" - }, - "bind_at": {"title": "绑定时间", "uitype": "text", "width": 150}, - "ext_json": {"title": "扩展JSON", "uitype": "textarea", "width": 200}, - "is_deleted": {"title": "逻辑删除", "uitype": "text", "width": 80}, - "creator_id": {"title": "创建人", "uitype": "text", "width": 110}, - "created_at": {"title": "创建时间", "uitype": "text", "width": 150}, - "updater_id": {"title": "更新人", "uitype": "text", "width": 110}, - "updated_at": {"title": "更新时间", "uitype": "text", "width": 150} + "exclouded": ["ext_json", "is_deleted", "updater_id", "updated_at"], + "alters": { + "ref_type": { + "uitype": "code", + "dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_domain_ref_type%27')}}" + }, + "bind_state": { + "uitype": "code", + "dataurl": "{{entire_url('/appbase/get_code.dspy?table=appcodes_kv&valuefield=k&textfield=v&cond=parentid%3D%27pbl_bind_state%27')}}" + }, + "blueprint_id": { + "uitype": "code", + "dataurl": "{{entire_url('/pbl_blueprint/pbl_blueprint_list')}}" + }, + "ext_json": { + "uitype": "textarea" + }, + "bind_at": { + "uitype": "text" + } + } }, "editexclouded": [ "id", "tenant_id", + "ref_type", + "ref_id", + "ref_code", + "ref_name", + "bind_at", + "is_deleted", "creator_id", "created_at", "updater_id", - "updated_at", - "is_deleted", - "bind_at" - ], - "exclouded": [ - "ext_json" - ], - "filters": ["tenant_id", "ref_type", "bind_state", "blueprint_id", "class_id", "team_id"], - "sortname": "bind_at", - "sortorder": "desc", - "rows": 20 + "updated_at" + ] } } diff --git a/pbl_domain_ext/api.py b/pbl_domain_ext/api.py index 639910d..1687057 100644 --- a/pbl_domain_ext/api.py +++ b/pbl_domain_ext/api.py @@ -209,11 +209,20 @@ def pbl_domain_ref_update(params): try: tenant_id = require_tenant(params) ref_pk = as_text(params.get("id")).strip() - if not ref_pk: - raise PblDomainExtError("PBL_DE_PARAM_INVALID", detail="id 为空") - exist = db.select_ref_by_id(tenant_id, ref_pk) + if ref_pk: + exist = db.select_ref_by_id(tenant_id, ref_pk) + else: + ref_type = check_ref_type(params.get("ref_type"), allow_empty=True) + ref_id = as_text(params.get("ref_id")).strip() + if not ref_type or not ref_id: + raise PblDomainExtError("PBL_DE_PARAM_INVALID", + detail="需 id 或 ref_type+ref_id") + exist = db.select_ref_by_key(tenant_id, ref_type, ref_id) if not exist: - raise PblDomainExtError("PBL_DE_NOT_FOUND", detail="id=%s" % ref_pk) + raise PblDomainExtError("PBL_DE_NOT_FOUND", + detail="id=%s ref_type=%s" % (ref_pk, + params.get("ref_type"))) + ref_pk = as_text(exist.get("id") or ref_pk).strip() row = {} for key in ("ref_code", "ref_name", "blueprint_id", "team_id", "class_id"): @@ -374,40 +383,49 @@ def pbl_world_list_by_tenant(params): def pbl_world_get_context(params): - """取世界上下文：world 只读视图 + 其下 scene/entity 关联 + PBL 绑定信息。""" + """只读聚合某 world 的上下文：world 本体 + 其下 scene/entity + PBL 关联。 + + 入参：tenant_id*, world_id* + 铁律：基表未命中或不属于本租户 → PBL_DE_NOT_FOUND（不得返回空壳成功体， + 否则跨租户探测会拿到 success=True 的假上下文）。 + """ try: tenant_id = require_tenant(params) - world_id = as_text(params.get("world_id") or params.get("ref_id") - or params.get("id")).strip() + world_id = as_text(params.get("world_id") or params.get("ref_id")).strip() if not world_id: raise PblDomainExtError("PBL_DE_PARAM_INVALID", detail="world_id 为空") - - base = db.read_base_row("world", world_id, tenant_id) - world_view = _base_view(base or {}) - ref = db.select_ref_by_key(tenant_id, "world", world_id) - - scenes = [_base_view(r) for r in db.read_base_rows("scene", tenant_id, - {"world_id": world_id}, limit=200)] - scene_ids = [as_text(s.get("id")) for s in scenes if s.get("id")] + raw = db.read_base_row("world", world_id, tenant_id) + if not raw: + raise PblDomainExtError("PBL_DE_NOT_FOUND", + detail="world.%s 不存在或不属于租户 %s" + % (world_id, tenant_id)) + world = _base_view(raw) + owner = as_text(world.get("tenant_id")).strip() + if owner and owner != tenant_id: + raise PblDomainExtError("PBL_DE_ACCESS_DENIED", + detail="world.%s 属他租户" % world_id) + if not as_text(world.get("id")).strip(): + world["id"] = world_id + scenes = (pbl_scene_list_by_world({"tenant_id": tenant_id, + "world_id": world_id} + ).get("data") or {}).get("items") or [] entities = [] - if params.get("with_entities") and scene_ids: - for raw in db.read_base_rows("entity", tenant_id, limit=500): - view = _base_view(raw) - if as_text(view.get("scene_id")) in scene_ids: - entities.append(view) - - return ok({ - "world": world_view, - "pbl_ref": _ref_out(ref) if ref else None, - "scenes": scenes, - "entities": entities, - "counts": {"scene": len(scenes), "entity": len(entities)}, - "readonly": True, - }) + for sc in scenes: + sid = as_text(sc.get("id")).strip() + if not sid: + continue + sub = pbl_entity_list_by_scene({"tenant_id": tenant_id, "scene_id": sid}) + entities.extend((sub.get("data") or {}).get("items") or []) + ref = db.select_ref_by_key(tenant_id, "world", world_id) + return ok({"world": world, + "pbl_ref": _ref_out(ref) if ref else None, + "scenes": scenes, + "entities": entities, + "counts": {"scene": len(scenes), "entity": len(entities)}, + "readonly": True}) except PblDomainExtError as exc: return exc.to_payload() - def pbl_scene_list_by_world(params): """只读列出某 world 下的 scene（基表按 world_id 过滤）+ 绑定标注。""" try: @@ -493,7 +511,11 @@ def pbl_team_bind_world(params): def pbl_team_world_list(params): - """列出团队（或班级）已绑定的 world。入参：tenant_id*, team_id 或 class_id。""" + """列出团队（或班级）已绑定的 world。入参：tenant_id*, team_id 或 class_id。 + + 返回 items 每项的 ``id``/``world_id`` 一律是**基表 world 主键**， + 关联表自身主键放 ``pbl_ref_id``——避免调用方把 ref 主键误当 world id。 + """ try: tenant_id = require_tenant(params) team_id = as_text(params.get("team_id")).strip() @@ -508,52 +530,56 @@ def pbl_team_world_list(params): conds["class_id"] = class_id rows = db.select_refs(tenant_id, conds=conds, limit=500) items = [] - for r in rows: - item = _ref_out(r) - item["world_id"] = item.get("ref_id") - item["world_code"] = item.get("ref_code") - item["world_name"] = item.get("ref_name") - items.append(item) - return ok({"items": items, "total": len(items), - "team_id": team_id, "class_id": class_id}) + seen = set() + for row in rows or []: + wid = as_text(row.get("ref_id")).strip() + if not wid or wid in seen: + continue + seen.add(wid) + view = _base_view(db.read_base_row("world", wid, tenant_id) or {}) + view["id"] = wid + view["world_id"] = wid + view["team_id"] = as_text(row.get("team_id")).strip() + view["class_id"] = as_text(row.get("class_id")).strip() + view["blueprint_id"] = as_text(row.get("blueprint_id")).strip() + view["bind_state"] = as_text(row.get("bind_state")).strip() + view["pbl_ref_id"] = as_text(row.get("id")).strip() + items.append(view) + return ok({"items": items, "total": len(items), "team_id": team_id, + "class_id": class_id, "ref_type": "world", "readonly": True}) except PblDomainExtError as exc: return exc.to_payload() - def pbl_team_list_by_class(params): - """按班级列出团队及其绑定世界数（聚合自 pbl_domain_ref，不新建表）。 - - 入参：tenant_id*, class_id* - 出参：data = {"items": [{"team_id", "class_id", "world_count", "worlds": [...]}]} - """ + """列出某班级下已绑定 world 的团队（按 team_id 去重）。入参：tenant_id*, class_id*。""" try: tenant_id = require_tenant(params) class_id = as_text(params.get("class_id")).strip() if not class_id: raise PblDomainExtError("PBL_DE_PARAM_INVALID", detail="class_id 为空") rows = db.select_refs(tenant_id, - conds={"class_id": class_id, "bind_state": "bound"}, - limit=500) - teams = {} - for r in rows: - team_id = as_text(r.get("team_id")).strip() or "(unassigned)" - bucket = teams.setdefault(team_id, {"team_id": team_id, - "class_id": class_id, - "world_count": 0, "worlds": []}) - if as_text(r.get("ref_type")) == "world": - bucket["world_count"] += 1 - bucket["worlds"].append({"world_id": r.get("ref_id"), - "code": r.get("ref_code"), - "name": r.get("ref_name")}) - items = sorted(teams.values(), key=lambda x: x["team_id"]) - return ok({"items": items, "total": len(items), "class_id": class_id}) + conds={"ref_type": "world", "class_id": class_id, + "bind_state": "bound"}, limit=500) + groups = {} + for row in rows or []: + tid = as_text(row.get("team_id")).strip() + key = tid or "__no_team__" + grp = groups.get(key) + if grp is None: + grp = {"team_id": tid, "class_id": class_id, + "world_ids": [], "world_count": 0} + groups[key] = grp + wid = as_text(row.get("ref_id")).strip() + if wid and wid not in grp["world_ids"]: + grp["world_ids"].append(wid) + for grp in groups.values(): + grp["world_count"] = len(grp["world_ids"]) + items = [groups[k] for k in sorted(groups)] + return ok({"items": items, "total": len(items), "class_id": class_id, + "readonly": True}) except PblDomainExtError as exc: return exc.to_payload() - -# -------------------------------------------------------------------------- -# 4. 编译期物化（pbl_compiler import 闭包所需符号） -# -------------------------------------------------------------------------- def pbl_domain_materialize_game_definition(params): """把蓝图关联的基础域对象物化为 Game Definition 片段（只读聚合，零写入）。 diff --git a/tests/fake_db.py b/tests/fake_db.py index 076b24f..0fa43bb 100644 --- a/tests/fake_db.py +++ b/tests/fake_db.py @@ -103,14 +103,27 @@ class FakeSor(object): self.sql_log.append(("C", tbl, dict(row))) return 1 - def U(self, tbl, row, where): + def U(self, tbl, ns, where=None): + """真实 sqlor 签名：``sor.U(table, ns)`` 两参，where 条件并入 ns。 + + 与真实实现同构：主键 ``id`` 与租户列 ``tenant_id`` 从 ns 中提出作 WHERE， + 其余列作 SET。三参调用（旧 fake 签名）仍兼容，便于对照回归。 + """ + ns = dict(ns or {}) + if where is None: + where = {} + for key in ("id", "tenant_id"): + if key in ns: + where[key] = ns.pop(key) + if not where: + where = {"id": ns.pop("id", None)} if "id" in ns else {} self._guard_readonly(tbl, "U") hits = 0 for exist in self.tables.get(tbl, []): if self._match(exist, where): - exist.update(row) + exist.update(ns) hits += 1 - self.sql_log.append(("U", tbl, dict(row), dict(where))) + self.sql_log.append(("U", tbl, dict(ns), dict(where))) return hits def D(self, tbl, where): diff --git a/tests/test_domain_ref.py b/tests/test_domain_ref.py index 978db5e..01eb627 100644 --- a/tests/test_domain_ref.py +++ b/tests/test_domain_ref.py @@ -80,7 +80,8 @@ class TestBind(BaseCase): self.assertEqual(str(data["ref_id"]), "W1") self.assertEqual(data["blueprint_id"], "BP1") self.assertEqual(data["bind_state"], "bound") - self.assertEqual(data["created_by"], OP) + self.assertEqual(data["creator_id"], OP, + "审计列名须与 models 权威名 creator_id 一致") self.assertEqual(len(self.rows()), 1, "应落 1 行关联记录") def test_bind_scene_and_entity_ok(self): @@ -152,7 +153,7 @@ class TestUnbindUpdateGet(BaseCase): self.assertIn("ext_json", data, "响应必须用设计 §J1 权威字段名 ext_json") self.assertEqual(self.rows()[0][EXT_FIELD] and '"a": 2' in self.rows()[0][EXT_FIELD] or '"a":2' in self.rows()[0][EXT_FIELD], True) - self.assertEqual(self.rows()[0]["updated_by"], OP) + self.assertEqual(self.rows()[0]["updater_id"], OP) def test_update_missing(self): self.assertErr(api.pbl_domain_ref_update( diff --git a/tests/test_models_contract.py b/tests/test_models_contract.py index 5ab0191..3826c7e 100644 --- a/tests/test_models_contract.py +++ b/tests/test_models_contract.py @@ -14,9 +14,12 @@ 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") @@ -126,43 +129,142 @@ class TestModelContract(unittest.TestCase): 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() + 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 与 models 字段名一致（QC#1 同步项）。""" + """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") - 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) + self.assertIn(key, self.params, "params 顶层缺 %s" % key) + self.assertIn("entire_url", self.params[key]) + self.assertIn("editable", self.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_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.crud["params"].get("editexclouded", []): - self.assertIn(col, self.model_fields) + 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_no_new_data_url_query_params(self): """new_data_url 不得带 query 参数（避免与 editexclouded 合并成 list）。""" - url = self.crud["params"]["new_data_url"] + 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 对齐。""" @@ -225,7 +327,10 @@ class TestIdGeneration(unittest.TestCase): 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) + # 平台 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) @@ -291,17 +396,19 @@ class TestSqlorSignatures(unittest.TestCase): 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"})) + # _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") diff --git a/ut.txt b/ut.txt deleted file mode 100644 index a6dfc43..0000000 --- a/ut.txt +++ /dev/null @@ -1,161 +0,0 @@ -......FF....F.......E...........F...../d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_bind.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_unbind.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_update.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_get.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_list.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_domain_ref_check_access.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_world_list_by_tenant.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_world_get_context.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_scene_list_by_world.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_entity_list_by_scene.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_team_bind_world.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_team_world_list.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py:425: ResourceWarning: unclosed file <_io.TextIOWrapper name='/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/wwwroot/api/pbl_team_list_by_class.dspy' mode='r' encoding='utf-8'> - body = open(path, encoding="utf-8").read() -ResourceWarning: Enable tracemalloc to get the object allocation traceback -.............FFFF....................F....E -====================================================================== -ERROR: test_bind_world_ok (test_domain_ref.TestBind) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 83, in test_bind_world_ok - self.assertEqual(data["created_by"], OP) -KeyError: 'created_by' - -====================================================================== -ERROR: test_update_ref_merges_where_into_ns (test_models_contract.TestSqlorSignatures) -update_ref 必须把 where 并入 ns 后两参调用 sor.U。 ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_models_contract.py", line 299, in test_update_ref_merges_where_into_ns - asyncio.get_event_loop().run_until_complete( - File "/usr/lib/python3.10/asyncio/base_events.py", line 628, in run_until_complete - future = tasks.ensure_future(future, loop=self) - File "/usr/lib/python3.10/asyncio/tasks.py", line 615, in ensure_future - return _ensure_future(coro_or_future, loop=loop) - File "/usr/lib/python3.10/asyncio/tasks.py", line 630, in _ensure_future - raise TypeError('An asyncio.Future, a coroutine or an awaitable ' -TypeError: An asyncio.Future, a coroutine or an awaitable is required - -====================================================================== -FAIL: test_world_get_context_cross_tenant (test_domain_ref.TestBaseDomainReadOnly) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 283, in test_world_get_context_cross_tenant - self.assertErr(api.pbl_world_get_context({"tenant_id": T2, "world_id": "W1"}), - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 52, in assertErr - self.assertFalse(resp.get("success"), "%s 期望失败，实得 %r" % (msg, resp)) -AssertionError: True is not false : W1 属 T1，T2 不得取其上下文 期望失败，实得 {'success': True, 'code': 'PBL_DE_OK', 'message': '成功', 'data': {'world': {}, 'pbl_ref': None, 'scenes': [], 'entities': [], 'counts': {'scene': 0, 'entity': 0}, 'readonly': True}} - -====================================================================== -FAIL: test_world_get_context_missing (test_domain_ref.TestBaseDomainReadOnly) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 279, in test_world_get_context_missing - self.assertErr(api.pbl_world_get_context({"tenant_id": T1, "world_id": "NOPE"}), - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 52, in assertErr - self.assertFalse(resp.get("success"), "%s 期望失败，实得 %r" % (msg, resp)) -AssertionError: True is not false : 期望失败，实得 {'success': True, 'code': 'PBL_DE_OK', 'message': '成功', 'data': {'world': {}, 'pbl_ref': None, 'scenes': [], 'entities': [], 'counts': {'scene': 0, 'entity': 0}, 'readonly': True}} - -====================================================================== -FAIL: test_bind_is_idempotent_upsert (test_domain_ref.TestBind) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 94, in test_bind_is_idempotent_upsert - self.assertOk(self.bind("world", "W1", blueprint_id="BP2")) - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 46, in assertOk - self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp)) -AssertionError: False is not true : 期望成功，实得 {'code': 'PBL_DE_DB_ERROR', 'message': '数据库操作失败', 'success': False, 'http_status': 500, 'detail': "update pbl_domain_ref: FakeSor.U() missing 1 required positional argument: 'where'"} - -====================================================================== -FAIL: test_team_world_list (test_domain_ref.TestTeamClass) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 348, in test_team_world_list - self.assertEqual(sorted(str(i.get("id") or i.get("world_id")) for i in items), ["W1"]) -AssertionError: Lists differ: ['9c7e0f0bf762efd9b64f8517cb8be84d'] != ['W1'] - -First differing element 0: -'9c7e0f0bf762efd9b64f8517cb8be84d' -'W1' - -- ['9c7e0f0bf762efd9b64f8517cb8be84d'] -+ ['W1'] - -====================================================================== -FAIL: test_unbind_soft_delete (test_domain_ref.TestUnbindUpdateGet) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 129, in test_unbind_soft_delete - data = self.assertOk(api.pbl_domain_ref_unbind( - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 46, in assertOk - self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp)) -AssertionError: False is not true : 期望成功，实得 {'code': 'PBL_DE_DB_ERROR', 'message': '数据库操作失败', 'success': False, 'http_status': 500, 'detail': "update pbl_domain_ref: FakeSor.U() missing 1 required positional argument: 'where'"} - -====================================================================== -FAIL: test_update_cross_tenant_denied (test_domain_ref.TestUnbindUpdateGet) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 164, in test_update_cross_tenant_denied - self.assertErr(api.pbl_domain_ref_update( - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 55, in assertErr - self.assertIn(resp.get("code"), codes, -AssertionError: 'PBL_DE_PARAM_INVALID' not found in ('PBL_DE_NOT_FOUND', 'PBL_DE_ACCESS_DENIED') : 跨租户不得改他租户关联 错误码期望 ['PBL_DE_NOT_FOUND', 'PBL_DE_ACCESS_DENIED']，实得 {'code': 'PBL_DE_PARAM_INVALID', 'message': '参数非法', 'success': False, 'http_status': 400, 'detail': 'id 为空'} - -====================================================================== -FAIL: test_update_ext_json_ok (test_domain_ref.TestUnbindUpdateGet) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 148, in test_update_ext_json_ok - data = self.assertOk(api.pbl_domain_ref_update( - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 46, in assertOk - self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp)) -AssertionError: False is not true : 期望成功，实得 {'code': 'PBL_DE_PARAM_INVALID', 'message': '参数非法', 'success': False, 'http_status': 400, 'detail': 'id 为空'} - -====================================================================== -FAIL: test_update_missing (test_domain_ref.TestUnbindUpdateGet) ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 158, in test_update_missing - self.assertErr(api.pbl_domain_ref_update( - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_domain_ref.py", line 55, in assertErr - self.assertIn(resp.get("code"), codes, -AssertionError: 'PBL_DE_PARAM_INVALID' not found in ('PBL_DE_NOT_FOUND',) : 更新不存在的关联 错误码期望 ['PBL_DE_NOT_FOUND']，实得 {'code': 'PBL_DE_PARAM_INVALID', 'message': '参数非法', 'success': False, 'http_status': 400, 'detail': 'id 为空'} - -====================================================================== -FAIL: test_sql_ddl_columns_match_model (test_models_contract.TestModelContract) -sql/pbl_domain_ext.sql 的建表列必须覆盖 models 全部列。 ----------------------------------------------------------------------- -Traceback (most recent call last): - File "/d/pipeline/workspaces/0/sdlc_general/modules/pbl_domain_ext/tests/test_models_contract.py", line 131, in test_sql_ddl_columns_match_model - self.assertRegex(ddl, r"`%s`" % col, "DDL 缺列 %s" % col) -AssertionError: Regex didn't match: '`ref_code`' not found in "-- =====================================================================\n-- pbl_domain_ext (M8) 基础域薄扩展 world/scene/entity\n-- 权威表定义: models/pbl_domain_ref.json (database-table-definition-spec 四段式)\n-- 字段名对齐 projects/pbls/docs/01-design/data-model.md §J1: ext_json LONGTEXT\n-- 约束: 仅新增关联表, 不改 world / scene / entity 三张基表结构 (Q-OPEN-3)\n-- 幂等: 全部 IF NOT EXISTS / ON DUPLICATE KEY UPDATE, 可重复执行\n-- =====================================================================\n\n-- ---------------------------------------------------------------------\n-- 1. 关联表 pbl_domain_ref (唯一新增表)\n-- ---------------------------------------------------------------------\nCREATE TABLE IF NOT EXISTS `pbl_domain_ref` (\n `id` VARCHAR(32) NOT NULL COMMENT '主键',\n `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(打头)',\n `ref_type` VARCHAR(32) NOT NULL COMMENT '关联对象类型(world/scene/entity)',\n `ref_id` BIGINT NOT NULL COMMENT '基表记录ID(world.id/scene.id/entity.id)',\n `blueprint_id` BIGINT DEFAULT NULL COMMENT '关联蓝图ID',\n `class_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '班级ID',\n `team_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '团队ID',\n `ext_json` LONGTEXT DEFAULT NULL COMMENT '扩展JSON(租户/班级/团队关联属性)',\n `created_by` VARCHAR(64) DEFAULT NULL COMMENT '创建人',\n `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',\n `updated_by` VARCHAR(64) DEFAULT NULL COMMENT '更新人',\n `updated_at` TIMESTAMP NULL DEFAULT NULL COMMENT '更新时间',\n PRIMARY KEY (`id`),\n UNIQUE KEY `uk_pbl_domain_ref` (`tenant_id`, `ref_type`, `ref_id`, `class_id`, `team_id`),\n KEY `idx_pbl_domain_ref_tenant_type` (`tenant_id`, `ref_type`),\n KEY `idx_pbl_domain_ref_blueprint` (`tenant_id`, `blueprint_id`),\n KEY `idx_pbl_domain_ref_class` (`tenant_id`, `class_id`),\n KEY `idx_pbl_domain_ref_team` (`tenant_id`, `team_id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL基础域薄扩展关联表(world/scene/entity)';\n\n-- ---------------------------------------------------------------------\n-- 2. 字典: pbl_domain_ref_type (models.codes 引用 appcodes_kv, cond=parentid=)\n-- appcodes 与 appcodes_kv 必须成对写入\n-- ---------------------------------------------------------------------\nINSERT INTO `appcodes` (`id`, `name`, `hierarchy_flg`) VALUES\n ('pbl_domain_ref_type', 'PBL关联对象类型', '0')\nON DUPLICATE KEY UPDATE `name` = VALUES(`name`);\n\nINSERT INTO `appcodes_kv` (`id`, `parentid`, `k`, `v`) VALUES\n ('pbl_drt_world', 'pbl_domain_ref_type', 'world', '世界'),\n ('pbl_drt_scene', 'pbl_domain_ref_type', 'scene', '场景'),\n ('pbl_drt_entity', 'pbl_domain_ref_type', 'entity', '实体')\nON DUPLICATE KEY UPDATE `v` = VALUES(`v`);\n\n-- ---------------------------------------------------------------------\n-- 3. 迁移兼容: 历史版本曾用列名 `ext`(json), 统一为设计权威名 ext_json(LONGTEXT)\n-- 存在旧列时搬迁数据后删除, 保证 models / sql / api 三处同名同型\n-- ---------------------------------------------------------------------\n-- (由部署脚本按 information_schema 判定后执行, 此处保留语句备查)\n-- SET @has_ext := (SELECT COUNT(*) FROM information_schema.COLUMNS\n-- WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'pbl_domain_ref' AND COLUMN_NAME = 'ext');\n-- SET @sql := IF(@has_ext > 0,\n-- 'UPDATE pbl_domain_ref SET ext_json = CAST(ext AS CHAR) WHERE ext_json IS NULL AND ext IS NOT NULL',\n-- 'SELECT 1');\n-- PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;\n-- SET @sql2 := IF(@has_ext > 0, 'ALTER TABLE pbl_domain_ref DROP COLUMN ext', 'SELECT 1');\n-- PREPARE s2 FROM @sql2; EXECUTE s2; DEALLOCATE PREPARE s2;\n\n-- ---------------------------------------------------------------------\n-- 4. 明确不做的事 (Q-OPEN-3 薄扩展边界)\n-- 不 ALTER world / scene / entity 三张基表; 不新增 world/scene/entity 三张 PBL 表;\n-- PBL 侧一律通过 pbl_domain_ref(ref_type + ref_id) 单向引用基表主键。\n-- ---------------------------------------------------------------------\n" : DDL 缺列 ref_code - ----------------------------------------------------------------------- -Ran 81 tests in 0.049s - -FAILED (failures=9, errors=2)