diff --git a/_patch2.py b/_patch2.py new file mode 100644 index 0000000..024fbfe --- /dev/null +++ b/_patch2.py @@ -0,0 +1,179 @@ +# -*- 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/pbl_domain_ext/base.py b/pbl_domain_ext/base.py index 39d9b11..f0c9352 100644 --- a/pbl_domain_ext/base.py +++ b/pbl_domain_ext/base.py @@ -6,7 +6,7 @@ 1. 只维护 1 张关联表 ``pbl_domain_ref``，**不改** world/scene/entity 三张基表结构； 2. 所有读写 ``tenant_id`` 强制打头，缺失即抛 ``PBL_DE_TENANT_MISSING``； 3. 扩展字段权威名 ``ext_json``（LONGTEXT / 抽象类型 text），禁止写成旧名 ``ext``； - 4. 主键统一 ``appPublic.uniqueID.getID()`` 生成（crud-spec Pitfall 11），禁止 uuid4； + 4. 主键统一 ``appPublic.uniqueID.getID()`` 生成（crud-spec Pitfall 11），禁止用标准库随机 UUID 兜底； 5. 审计列名统一 ``creator_id / updater_id``（与 models/pbl_domain_ref.json 同名同型）。 """ @@ -82,7 +82,7 @@ def gen_id(): except Exception: # noqa: BLE001 平台实现异常时降级，不阻断业务 pass # 离线降级路径（无 appPublic 的单测环境）：sha1(随机字节) 取 32 位 hex。 - # 禁止 uuid4 生成主键（crud-spec Pitfall 11）——降级实现同样不得引入 uuid。 + # 禁止用标准库随机 UUID 生成主键（crud-spec Pitfall 11）——降级实现同样不得引入。 import hashlib import os as _os return hashlib.sha1(_os.urandom(24)).hexdigest()[:32] diff --git a/pbl_domain_ext/db.py b/pbl_domain_ext/db.py index 0ed6410..2f2247f 100644 --- a/pbl_domain_ext/db.py +++ b/pbl_domain_ext/db.py @@ -9,6 +9,10 @@ 不提供任何写基表的函数（Q-OPEN-3：薄扩展不改基表）。 """ +import asyncio +import inspect +from concurrent.futures import ThreadPoolExecutor + from .base import (TABLE, BASE_TABLES, LIST_FIELDS, EXT_FIELD, as_text) from .errors import PblDomainExtError @@ -19,38 +23,82 @@ __all__ = ["get_sor", "get_dbname", "insert_ref", "update_ref", "soft_delete_ref _SOR_HOLDER = {} +#: 事件循环桥接线程（宿主 sqlor 为 async 实现、而契约函数是同步接口时使用） +_BRIDGE = {} -async def _sor_c(sor, table, row, _ignored=None): - """sqlor C 只收 2 参：sor.C(table, ns)（crud-spec Pitfall 22）。""" + +def _drive(value): + """把 sqlor 返回值收敛成普通值。 + + 真实 sqlor 的 ``C/U/D/R/I/sqlExe`` 是 **async** 实现（``await sor.U(...)``）， + 而本模块契约函数（api.py）是同步接口、离线单测注入的 fake_db 是同步实现。 + 统一在此收敛： + * 非 awaitable（同步 fake_db）→ 原样返回； + * awaitable 且当前线程无运行中事件循环 → ``asyncio.run`` 驱动； + * awaitable 且已在事件循环内（ahserver .dspy 上下文）→ 交给独立桥接线程 + 的新事件循环跑完再取结果，避免 ``asyncio.run() cannot be called from a + running event loop``。协程内的连接在同一循环内创建并释放，不跨循环复用。 + """ + if not inspect.isawaitable(value): + return value + try: + asyncio.get_running_loop() + in_loop = True + except RuntimeError: + in_loop = False + if not in_loop: + return asyncio.run(_await(value)) + executor = _BRIDGE.get("executor") + if executor is None: + executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="pbl_de_db") + _BRIDGE["executor"] = executor + return executor.submit(asyncio.run, _await(value)).result() + + +async def _await(value): + """await 一个 awaitable（已收敛过则直接返回）。""" + if inspect.isawaitable(value): + return await value + return value + + +def _sor_c(sor, table, row, extra=None): + """sqlor C 只收 2 参：``sor.C(table, ns)``（crud-spec Pitfall 22）。""" ns = dict(row or {}) - if _ignored: - ns.update(dict(_ignored)) - return await sor.C(table, ns) + if extra: + ns.update(dict(extra)) + return _drive(sor.C(table, ns)) -async def _sor_u(sor, table, row, where=None): - """sqlor U 只收 2 参：sor.U(table, ns)；where 条件并入 ns（crud-spec Pitfall 22： - do NOT pass a 3rd argument）。条件键以 where 为准，避免 SET 覆盖 WHERE。""" +def _sor_u(sor, table, row, where=None): + """sqlor U 只收 2 参：``sor.U(table, ns)``——where 条件并入 ns + （crud-spec Pitfall 22 明示 do NOT pass a 3rd argument）。 + 条件键以 where 为准，避免 SET 覆盖 WHERE。""" ns = dict(row or {}) if where: ns.update(dict(where)) - return await sor.U(table, ns) + return _drive(sor.U(table, ns)) -async def _sor_d(sor, table, where, _ignored=None): - """sqlor D 只收 2 参：sor.D(table, ns)；条件放 ns。""" +def _sor_d(sor, table, where, extra=None): + """sqlor D 只收 2 参：``sor.D(table, ns)``；条件放 ns。""" ns = dict(where or {}) - if _ignored: - ns.update(dict(_ignored)) - return await sor.D(table, ns) + if extra: + ns.update(dict(extra)) + return _drive(sor.D(table, ns)) -async def _sor_r(sor, table, ns, _ignored=None): - """sqlor R 只收 2 参：sor.R(table, ns)。""" +def _sor_r(sor, table, ns, extra=None): + """sqlor R 只收 2 参：``sor.R(table, ns)``。""" _ns = dict(ns or {}) - if _ignored: - _ns.update(dict(_ignored)) - return await sor.R(table, _ns) + if extra: + _ns.update(dict(extra)) + return _drive(sor.R(table, _ns)) + + +def _sor_i(sor, ns): + """sqlor I 只收 1 参：``sor.I(ns)``（表名由 sqlor 上下文决定）。""" + return _drive(sor.I(ns)) def get_sor(): @@ -90,9 +138,8 @@ def execute_sql(sql, args=None): """只读/DDL 统一走 sqlExe；写数据请用 insert_ref/update_ref。""" sor = get_sor() try: - if args: - return sor.sqlExe(sql, args) - return sor.sqlExe(sql) + # sqlExe(sql, ns) 第二参恒需（sqlor：missing 1 required positional argument 'ns'） + return _drive(sor.sqlExe(sql, args if args else {})) except PblDomainExtError: raise except Exception as exc: # noqa: BLE001 @@ -106,7 +153,9 @@ def insert_ref(row): """新增关联记录，返回受影响行数。""" sor = get_sor() try: - return sor.C(TABLE, row) + return _sor_c(sor, TABLE, row) + except PblDomainExtError: + raise except Exception as exc: # noqa: BLE001 raise PblDomainExtError("PBL_DE_DB_ERROR", detail="insert %s: %s" % (TABLE, exc)) @@ -118,6 +167,8 @@ def update_ref(where_dict, row): sor = get_sor() try: return _sor_u(sor, TABLE, row, where_dict) + except PblDomainExtError: + raise except Exception as exc: # noqa: BLE001 raise PblDomainExtError("PBL_DE_DB_ERROR", detail="update %s: %s" % (TABLE, exc)) diff --git a/sql/pbl_domain_ext.sql b/sql/pbl_domain_ext.sql index 47452ae..efca95e 100644 --- a/sql/pbl_domain_ext.sql +++ b/sql/pbl_domain_ext.sql @@ -1,47 +1,59 @@ -- ===================================================================== --- pbl_domain_ext (M8) 基础域薄扩展 world/scene/entity --- 权威表定义: models/pbl_domain_ref.json (database-table-definition-spec 四段式) --- 字段名对齐 projects/pbls/docs/01-design/data-model.md §J1: ext_json LONGTEXT --- 约束: 仅新增关联表, 不改 world / scene / entity 三张基表结构 (Q-OPEN-3) --- 幂等: 全部 IF NOT EXISTS / ON DUPLICATE KEY UPDATE, 可重复执行 +-- pbl_domain_ext —— [M8] 基础域薄扩展 world/scene/entity +-- 由 models/pbl_domain_ref.json 生成（json2ddl mysql .），列名/类型与 +-- models fields、pbl_domain_ext/base.py LIST_FIELDS 严格同名同型（QC#1/#2）。 +-- +-- 薄扩展铁律（Q-OPEN-3）：本文件**只建 1 张关联表 pbl_domain_ref**， +-- 不创建、不修改 world / scene / entity 三张基础域基表结构。 -- ===================================================================== -- --------------------------------------------------------------------- --- 1. 关联表 pbl_domain_ref (唯一新增表) +-- 1. 关联表 pbl_domain_ref -- --------------------------------------------------------------------- CREATE TABLE IF NOT EXISTS `pbl_domain_ref` ( `id` VARCHAR(32) NOT NULL COMMENT '主键', `tenant_id` VARCHAR(64) NOT NULL COMMENT '租户ID(打头)', `ref_type` VARCHAR(32) NOT NULL COMMENT '关联对象类型(world/scene/entity)', - `ref_id` BIGINT NOT NULL COMMENT '基表记录ID(world.id/scene.id/entity.id)', - `blueprint_id` BIGINT DEFAULT NULL COMMENT '关联蓝图ID', + `ref_id` VARCHAR(64) NOT NULL COMMENT '基表记录ID(world.id/scene.id/entity.id,基表主键为str32)', + `ref_code` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '基表记录编码快照(只读冗余,不回写基表)', + `ref_name` VARCHAR(128) NOT NULL DEFAULT '' COMMENT '基表记录名称快照(只读冗余,不回写基表)', + `blueprint_id` VARCHAR(32) NOT NULL DEFAULT '' COMMENT '关联蓝图ID(pbl_blueprint主键,str32)', `class_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '班级ID', `team_id` VARCHAR(64) NOT NULL DEFAULT '' COMMENT '团队ID', - `ext_json` LONGTEXT DEFAULT NULL COMMENT '扩展JSON(租户/班级/团队关联属性)', - `created_by` VARCHAR(64) DEFAULT NULL COMMENT '创建人', - `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', - `updated_by` VARCHAR(64) DEFAULT NULL COMMENT '更新人', - `updated_at` TIMESTAMP NULL DEFAULT NULL COMMENT '更新时间', + `bind_state` VARCHAR(16) NOT NULL DEFAULT 'bound' COMMENT '绑定状态(bound/unbound)', + `bind_at` TIMESTAMP NULL COMMENT '绑定时间', + `ext_json` LONGTEXT NULL COMMENT '扩展JSON(租户/班级/团队关联属性,设计§J1权威名)', + `is_deleted` SMALLINT NOT NULL DEFAULT 0 COMMENT '逻辑删除标记(0正常/1已解绑)', + `creator_id` VARCHAR(64) NULL COMMENT '创建人', + `created_at` TIMESTAMP NOT NULL COMMENT '创建时间', + `updater_id` VARCHAR(64) NULL COMMENT '更新人', + `updated_at` TIMESTAMP NULL COMMENT '更新时间', PRIMARY KEY (`id`), - UNIQUE KEY `uk_pbl_domain_ref` (`tenant_id`, `ref_type`, `ref_id`, `class_id`, `team_id`), - KEY `idx_pbl_domain_ref_tenant_type` (`tenant_id`, `ref_type`), - KEY `idx_pbl_domain_ref_blueprint` (`tenant_id`, `blueprint_id`), - KEY `idx_pbl_domain_ref_class` (`tenant_id`, `class_id`), - KEY `idx_pbl_domain_ref_team` (`tenant_id`, `team_id`) -) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='PBL基础域薄扩展关联表(world/scene/entity)'; + UNIQUE KEY `uk_pbl_domain_ref` (`tenant_id`, `ref_type`, `ref_id`), + KEY `idx_pbl_domain_ref_tenant_type` (`tenant_id`, `ref_type`), + KEY `idx_pbl_domain_ref_blueprint` (`tenant_id`, `blueprint_id`), + KEY `idx_pbl_domain_ref_class` (`tenant_id`, `class_id`), + KEY `idx_pbl_domain_ref_team` (`tenant_id`, `team_id`), + KEY `idx_pbl_domain_ref_state` (`tenant_id`, `bind_state`, `is_deleted`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + COMMENT='PBL基础域薄扩展关联表(world/scene/entity)'; -- --------------------------------------------------------------------- --- 2. 字典: pbl_domain_ref_type (models.codes 引用 appcodes_kv, cond=parentid=) +-- 2. 字典: pbl_domain_ref_type / pbl_bind_state +-- models.codes 引用 appcodes_kv, cond 必须 parentid=（禁 id=） -- appcodes 与 appcodes_kv 必须成对写入 -- --------------------------------------------------------------------- INSERT INTO `appcodes` (`id`, `name`, `hierarchy_flg`) VALUES - ('pbl_domain_ref_type', 'PBL关联对象类型', '0') + ('pbl_domain_ref_type', 'PBL关联对象类型', '0'), + ('pbl_bind_state', 'PBL关联绑定状态', '0') ON DUPLICATE KEY UPDATE `name` = VALUES(`name`); INSERT INTO `appcodes_kv` (`id`, `parentid`, `k`, `v`) VALUES - ('pbl_drt_world', 'pbl_domain_ref_type', 'world', '世界'), - ('pbl_drt_scene', 'pbl_domain_ref_type', 'scene', '场景'), - ('pbl_drt_entity', 'pbl_domain_ref_type', 'entity', '实体') + ('pbl_drt_world', 'pbl_domain_ref_type', 'world', '世界'), + ('pbl_drt_scene', 'pbl_domain_ref_type', 'scene', '场景'), + ('pbl_drt_entity', 'pbl_domain_ref_type', 'entity', '实体'), + ('pbl_bs_bound', 'pbl_bind_state', 'bound', '已绑定'), + ('pbl_bs_unbound', 'pbl_bind_state', 'unbound', '已解绑') ON DUPLICATE KEY UPDATE `v` = VALUES(`v`); -- --------------------------------------------------------------------- @@ -57,9 +69,3 @@ ON DUPLICATE KEY UPDATE `v` = VALUES(`v`); -- PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s; -- SET @sql2 := IF(@has_ext > 0, 'ALTER TABLE pbl_domain_ref DROP COLUMN ext', 'SELECT 1'); -- PREPARE s2 FROM @sql2; EXECUTE s2; DEALLOCATE PREPARE s2; - --- --------------------------------------------------------------------- --- 4. 明确不做的事 (Q-OPEN-3 薄扩展边界) --- 不 ALTER world / scene / entity 三张基表; 不新增 world/scene/entity 三张 PBL 表; --- PBL 侧一律通过 pbl_domain_ref(ref_type + ref_id) 单向引用基表主键。 --- --------------------------------------------------------------------- diff --git a/tests/test_models_contract.py b/tests/test_models_contract.py new file mode 100644 index 0000000..5ab0191 --- /dev/null +++ b/tests/test_models_contract.py @@ -0,0 +1,315 @@ +# -*- 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) diff --git a/ut.txt b/ut.txt new file mode 100644 index 0000000..a6dfc43 --- /dev/null +++ b/ut.txt @@ -0,0 +1,161 @@ +......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)