diff --git a/json/pbl_domain_ref.json b/json/pbl_domain_ref.json index 341a7eb..c1a0716 100644 --- a/json/pbl_domain_ref.json +++ b/json/pbl_domain_ref.json @@ -78,6 +78,9 @@ "new_data_url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_bind.dspy')}}", "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')}}" - } + }, + "new_data_url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_bind.dspy')}}", + "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')}}" } -} +} \ No newline at end of file diff --git a/pbl_domain_ext/init.py b/pbl_domain_ext/init.py index 7845c06..cb9cb8b 100644 --- a/pbl_domain_ext/init.py +++ b/pbl_domain_ext/init.py @@ -15,6 +15,7 @@ from .base import TABLE, EXT_FIELD, REF_TYPES, BIND_STATES from .errors import PblDomainExtError __all__ = ["load_pbl_domain_ext", "MODULE_NAME", "API_ROUTES", "PERMISSIONS", + "CONTRACT_FUNCS", "_register_serverenv", "TABLE", "EXT_FIELD", "REF_TYPES", "BIND_STATES"] MODULE_NAME = "pbl_domain_ext" @@ -47,6 +48,46 @@ PERMISSIONS = [ ] +#: 契约函数名（与 wwwroot/api/*.dspy 一一对应；三处注册之③ load 期挂 ServerEnv） +CONTRACT_FUNCS = [ + "pbl_domain_ref_bind", "pbl_domain_ref_unbind", "pbl_domain_ref_update", + "pbl_domain_ref_get", "pbl_domain_ref_list", "pbl_domain_ref_check_access", + "pbl_world_list_by_tenant", "pbl_world_get_context", + "pbl_scene_list_by_world", "pbl_entity_list_by_scene", + "pbl_team_bind_world", "pbl_team_world_list", "pbl_team_list_by_class", + "pbl_domain_materialize_game_definition", +] + + +def _register_serverenv(): + """把契约函数注册为 ServerEnv 全局（.dspy 直接按名调用，无需 import）。 + + module-development-spec「Triple-place function registration」之第③处： + ① 实现 pbl_domain_ext/api.py ② 包导出 pbl_domain_ext/__init__.py + ③ 本函数 env. = + 缺③ → dspy 运行期 NameError: name 'pbl_xxx' is not defined。 + """ + registered, missing = [], [] + env = None + try: + from ahserver.serverenv import ServerEnv + env = ServerEnv() + except Exception: # noqa: BLE001 非 ahserver 宿主（单测/离线）跳过 + env = None + for name in CONTRACT_FUNCS: + func = getattr(api, name, None) + if func is None: + missing.append(name) + continue + if env is not None: + try: + setattr(env, name, func) + except Exception: # noqa: BLE001 + pass + registered.append(name) + return {"registered": registered, "missing": missing, "serverenv": env is not None} + + def _register_routes(app=None): """注册 dspy 路由（平台 register_dspy 可用时走平台，否则仅返回路由表）。""" registered = [] @@ -83,6 +124,7 @@ def load_pbl_domain_ext(app=None, sor=None, ensure_schema=True): """ if sor is not None: db.set_sor(sor) + env_info = _register_serverenv() routes = _register_routes(app) perms = _register_permissions(app) @@ -100,6 +142,7 @@ def load_pbl_domain_ext(app=None, sor=None, ensure_schema=True): "ref_types": list(REF_TYPES), "bind_states": list(BIND_STATES), "routes": routes, + "serverenv": env_info, "permissions": perms, "schema_action": schema_action, "base_tables_readonly": ["world", "scene", "entity"], diff --git a/tests/test_domain_ref.py b/tests/test_domain_ref.py index 8423858..978db5e 100644 --- a/tests/test_domain_ref.py +++ b/tests/test_domain_ref.py @@ -1,528 +1,448 @@ -"""tests/test_domain_ref.py — M8 薄扩展 13 个契约接口的真实断言测试（离线 sqlite）。 +# -*- coding: utf-8 -*- +"""[M8] pbl_domain_ext 契约测试（world/scene/entity 薄扩展）。 -覆盖： - §3.1 bind_ref / unbind_ref / get_ref / list_refs / update_ref - §3.2 list_worlds_by_tenant / list_scenes_by_world / list_entities_by_scene - / get_world_with_pbl_context / check_ref_access - §3.3 list_teams_by_class / bind_team_to_world / get_team_worlds - + 薄扩展铁律：不改基表（基表行数/结构前后一致）、跨租户隔离（US-21）、 - 悬挂引用过滤、表总账（OWN_TABLES == models/*.json == 1 张 pbl_domain_ref）。 +运行（环境无 pytest 时用 unittest，二者皆可）:: -运行：python3 tests/test_domain_ref.py + cd modules/pbl_domain_ext + python3 -m unittest discover -s tests -p 'test_*.py' -v + python3 -m pytest tests/ -q # 若环境已装 pytest + +覆盖：13 个 dspy 契约对应的 api 函数 × 租户隔离 / 权限(只读基表) / 正常 / 异常 四类用例， +外加薄扩展铁律（不写基表、表总账只有 pbl_domain_ref 一张新表、ext_json 字段名对齐设计 §J1）。 """ -import asyncio -import json import os import sys import unittest HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) -sys.path.insert(0, ROOT) -sys.path.insert(0, HERE) +for p in (ROOT, HERE): + if p not in sys.path: + sys.path.insert(0, p) -import fake_db # noqa: E402 +import fake_db # noqa: E402 +from pbl_domain_ext import api, db, base # noqa: E402 +from pbl_domain_ext.base import EXT_FIELD, REF_TYPES, TABLE # noqa: E402 -from pbl_domain_ext import api # noqa: E402 -from pbl_domain_ext import base as base_mod # noqa: E402 -from pbl_domain_ext import init as init_mod # noqa: E402 -from pbl_domain_ext.errors import (E_DUPLICATE, E_FORBIDDEN, E_NOT_FOUND, # noqa: E402 - E_VALIDATION, PblError) - -T1 = 'T1' -T2 = 'T2' - - -def run(coro): - return asyncio.get_event_loop().run_until_complete(coro) if False else asyncio.run(coro) +T1 = "T1" # 主测租户（BASE_FIXTURES 中 world W1/W2、scene S1..S3、entity E1..E3） +T2 = "T2" # 他租户（BASE_FIXTURES 中仅 world W3） +OP = "U_TESTER" class BaseCase(unittest.TestCase): + """每个用例独立内存库，互不污染。""" + def setUp(self): - self.conn, self.adapter = fake_db.setup() + self.sor = fake_db.make_fake_sor() + db.set_sor(self.sor) def tearDown(self): - fake_db.teardown() - self.conn.close() + db.set_sor(None) - def assertPblError(self, code, coro): - with self.assertRaises(Exception) as ctx: - run(coro) - exc = ctx.exception - self.assertEqual(getattr(exc, 'code', None), code, - '期望 %s，实际 %s(%s)' % (code, getattr(exc, 'code', None), exc)) - return exc + # ---- 断言助手 ------------------------------------------------------- + def assertOk(self, resp, msg=""): + self.assertIsInstance(resp, dict, "契约必须返回 dict 响应体") + self.assertTrue(resp.get("success"), "%s 期望成功，实得 %r" % (msg, resp)) + self.assertEqual(resp.get("code"), "PBL_DE_OK", msg) + return resp.get("data") + + def assertErr(self, resp, codes, msg=""): + self.assertIsInstance(resp, dict, "契约必须返回 dict 响应体（不得裸抛栈）") + self.assertFalse(resp.get("success"), "%s 期望失败，实得 %r" % (msg, resp)) + if isinstance(codes, str): + codes = (codes,) + self.assertIn(resp.get("code"), codes, + "%s 错误码期望 %s，实得 %r" % (msg, list(codes), resp)) + self.assertIn("http_status", resp, "失败响应必须带 http_status") + return resp + + def bind(self, ref_type, ref_id, tenant_id=T1, **kw): + params = {"tenant_id": tenant_id, "ref_type": ref_type, + "ref_id": ref_id, "operator_id": OP} + params.update(kw) + return api.pbl_domain_ref_bind(params) + + def rows(self, table=TABLE): + return list(self.sor.tables.get(table, [])) -class TestBindRef(BaseCase): - """接口 1：bind_ref""" +# ========================================================================== +# 1. 关联表 pbl_domain_ref：绑定 / 解绑 / 更新 / 查询 +# ========================================================================== +class TestBind(BaseCase): def test_bind_world_ok(self): - ref = run(api.bind_ref('world', 1, blueprint_id='BP-1', class_id='CLS-1', - team_id='TEAM_A', ext_json={'stage': 2}, tenant_id=T1)) - self.assertIsNotNone(ref) - self.assertEqual(ref['tenant_id'], T1) - self.assertEqual(ref['ref_type'], 'world') - self.assertEqual(ref['ref_id'], 1) - self.assertEqual(ref['blueprint_id'], 'BP-1') - self.assertEqual(ref['class_id'], 'CLS-1') - self.assertEqual(ref['team_id'], 'TEAM_A') - self.assertEqual(ref['ext_json'], {'stage': 2}) - self.assertEqual(ref['is_deleted'], 0) + data = self.assertOk(self.bind("world", "W1", blueprint_id="BP1", + class_id="C1", team_id="TEAM_A", + ext_json='{"role":"host"}')) + self.assertEqual(data["ref_type"], "world") + 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(len(self.rows()), 1, "应落 1 行关联记录") - def test_bind_duplicate_raises(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.assertPblError(E_DUPLICATE, api.bind_ref('world', 1, tenant_id=T1)) + def test_bind_scene_and_entity_ok(self): + self.assertOk(self.bind("scene", "S1")) + self.assertOk(self.bind("entity", "E1")) + self.assertEqual(sorted(r["ref_type"] for r in self.rows()), + ["entity", "scene"]) - def test_bind_same_ref_other_tenant_ok(self): - """同一基表记录可被不同租户各自绑定（UNIQUE 含 tenant_id）。""" - run(api.bind_ref('world', 1, tenant_id=T1)) - ref2 = run(api.bind_ref('world', 1, tenant_id=T2)) - self.assertEqual(ref2['tenant_id'], T2) + def test_bind_is_idempotent_upsert(self): + self.assertOk(self.bind("world", "W1", blueprint_id="BP1")) + self.assertOk(self.bind("world", "W1", blueprint_id="BP2")) + self.assertEqual(len(self.rows()), 1, "同键重复绑定不得产生重复行（幂等 upsert）") + self.assertEqual(self.rows()[0]["blueprint_id"], "BP2") - def test_bind_base_missing_raises_not_found(self): - self.assertPblError(E_NOT_FOUND, api.bind_ref('world', 999, tenant_id=T1)) + def test_bind_requires_tenant(self): + self.assertErr(api.pbl_domain_ref_bind({"ref_type": "world", "ref_id": "W1"}), + "PBL_DE_TENANT_MISSING", "缺 tenant_id") - def test_bind_invalid_ref_type(self): - self.assertPblError(E_VALIDATION, api.bind_ref('script', 1, tenant_id=T1)) + def test_bind_rejects_bad_ref_type(self): + self.assertErr(self.bind("course", "W1"), "PBL_DE_REF_TYPE_INVALID", + "ref_type 仅允许 world/scene/entity") - def test_bind_invalid_ref_id(self): - self.assertPblError(E_VALIDATION, api.bind_ref('world', 'abc', tenant_id=T1)) - self.assertPblError(E_VALIDATION, api.bind_ref('world', 0, tenant_id=T1)) + def test_bind_rejects_empty_ref_id(self): + self.assertErr(self.bind("world", ""), "PBL_DE_PARAM_INVALID", "ref_id 为空") - def test_bind_missing_tenant(self): - self.assertPblError(E_VALIDATION, api.bind_ref('world', 1)) + def test_bind_rejects_missing_base_row(self): + self.assertErr(self.bind("world", "W_NOT_EXIST"), + ("PBL_DE_BASE_MISSING", "PBL_DE_NOT_FOUND"), + "基表未命中的对象不得建立关联") - def test_rebind_after_unbind_revives(self): - run(api.bind_ref('scene', 11, class_id='CLS-1', tenant_id=T1)) - run(api.unbind_ref('scene', 11, tenant_id=T1)) - ref = run(api.bind_ref('scene', 11, class_id='CLS-2', tenant_id=T1)) - self.assertEqual(ref['class_id'], 'CLS-2') - self.assertEqual(ref['is_deleted'], 0) + def test_bind_rejects_invalid_ext_json(self): + self.assertErr(self.bind("world", "W1", ext_json="{not json"), + "PBL_DE_EXT_JSON_INVALID", "ext_json 必须合法 JSON") - def test_bind_ext_invalid_json_string(self): - self.assertPblError(E_VALIDATION, - api.bind_ref('world', 1, ext_json='{bad json', tenant_id=T1)) + def test_bind_tenant_isolation(self): + self.assertOk(self.bind("world", "W1", tenant_id=T1)) + self.assertErr(self.bind("world", "W1", tenant_id=T2), + ("PBL_DE_BASE_MISSING", "PBL_DE_NOT_FOUND", "PBL_DE_ACCESS_DENIED"), + "W1 属 T1，T2 不得绑定") -class TestUnbindGetUpdate(BaseCase): - """接口 2/3/5：unbind_ref / get_ref / update_ref""" +class TestUnbindUpdateGet(BaseCase): def test_unbind_soft_delete(self): - run(api.bind_ref('entity', 101, tenant_id=T1)) - self.assertTrue(run(api.unbind_ref('entity', 101, tenant_id=T1))) - flag = fake_db.scalar(self.conn, - "SELECT is_deleted FROM pbl_domain_ref WHERE ref_id=101") - self.assertEqual(flag, 1, 'unbind 必须软删而非物理删除') - self.assertPblError(E_NOT_FOUND, api.get_ref('entity', 101, tenant_id=T1)) + self.assertOk(self.bind("world", "W1")) + data = self.assertOk(api.pbl_domain_ref_unbind( + {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "operator_id": OP})) + self.assertIn(data.get("bind_state", "unbound"), ("unbound", "deleted")) + alive = [r for r in self.rows() if str(r.get("is_deleted", "0")) in ("0", "")] + self.assertEqual(alive, [], "解绑后不应再有生效关联行") def test_unbind_missing(self): - self.assertPblError(E_NOT_FOUND, api.unbind_ref('world', 1, tenant_id=T1)) + self.assertErr(api.pbl_domain_ref_unbind( + {"tenant_id": T1, "ref_type": "world", "ref_id": "W1"}), + "PBL_DE_NOT_FOUND", "解绑不存在的关联") - def test_get_ref_ok(self): - run(api.bind_ref('world', 2, blueprint_id='BP-9', tenant_id=T1)) - ref = run(api.get_ref('world', 2, tenant_id=T1)) - self.assertEqual(ref['blueprint_id'], 'BP-9') + def test_unbind_cross_tenant_denied(self): + self.assertOk(self.bind("world", "W1", tenant_id=T1)) + self.assertErr(api.pbl_domain_ref_unbind( + {"tenant_id": T2, "ref_type": "world", "ref_id": "W1"}), + ("PBL_DE_NOT_FOUND", "PBL_DE_ACCESS_DENIED"), "跨租户不得解绑他租户关联") - def test_get_ref_cross_tenant_not_found(self): - run(api.bind_ref('world', 2, tenant_id=T1)) - self.assertPblError(E_NOT_FOUND, api.get_ref('world', 2, tenant_id=T2)) + def test_update_ext_json_ok(self): + self.assertOk(self.bind("world", "W1", ext_json='{"a":1}')) + data = self.assertOk(api.pbl_domain_ref_update( + {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", + "ext_json": '{"a":2,"b":"x"}', "class_id": "C9", "operator_id": OP})) + self.assertEqual(data["class_id"], "C9") + 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) - def test_update_ref_ok(self): - run(api.bind_ref('world', 1, class_id='CLS-1', tenant_id=T1)) - ref = run(api.update_ref('world', 1, {'class_id': 'CLS-2', 'team_id': 'TEAM_B', - 'ext_json': {'k': 'v'}}, tenant_id=T1)) - self.assertEqual(ref['class_id'], 'CLS-2') - self.assertEqual(ref['team_id'], 'TEAM_B') - self.assertEqual(ref['ext_json'], {'k': 'v'}) + def test_update_missing(self): + self.assertErr(api.pbl_domain_ref_update( + {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "class_id": "C1"}), + "PBL_DE_NOT_FOUND", "更新不存在的关联") - def test_update_ref_rejects_key_fields(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.assertPblError(E_VALIDATION, - api.update_ref('world', 1, {'ref_id': 2}, tenant_id=T1)) - self.assertPblError(E_VALIDATION, - api.update_ref('world', 1, {'tenant_id': T2}, tenant_id=T1)) + def test_update_cross_tenant_denied(self): + self.assertOk(self.bind("world", "W1", tenant_id=T1)) + self.assertErr(api.pbl_domain_ref_update( + {"tenant_id": T2, "ref_type": "world", "ref_id": "W1", "class_id": "X"}), + ("PBL_DE_NOT_FOUND", "PBL_DE_ACCESS_DENIED"), "跨租户不得改他租户关联") - def test_update_ref_missing(self): - self.assertPblError(E_NOT_FOUND, - api.update_ref('world', 1, {'class_id': 'X'}, tenant_id=T1)) + def test_get_by_id_and_by_key(self): + bound = self.assertOk(self.bind("scene", "S1", blueprint_id="BP7")) + by_id = self.assertOk(api.pbl_domain_ref_get({"tenant_id": T1, "id": bound["id"]})) + self.assertEqual(by_id["id"], bound["id"]) + by_key = self.assertOk(api.pbl_domain_ref_get( + {"tenant_id": T1, "ref_type": "scene", "ref_id": "S1", "with_base": 1})) + self.assertEqual(by_key["id"], bound["id"]) + self.assertEqual(by_key["base"]["id"], "S1", "with_base 应附基表只读视图") + + def test_get_missing(self): + self.assertErr(api.pbl_domain_ref_get({"tenant_id": T1, "id": "NOPE"}), + "PBL_DE_NOT_FOUND") + + def test_get_cross_tenant_denied(self): + bound = self.assertOk(self.bind("world", "W1", tenant_id=T1)) + self.assertErr(api.pbl_domain_ref_get({"tenant_id": T2, "id": bound["id"]}), + ("PBL_DE_NOT_FOUND", "PBL_DE_ACCESS_DENIED"), + "T2 不得读 T1 的关联记录") + + def test_list_filter_and_paging(self): + self.assertOk(self.bind("world", "W1", class_id="C1", team_id="TEAM_A")) + self.assertOk(self.bind("scene", "S1", class_id="C1", team_id="TEAM_A")) + self.assertOk(self.bind("entity", "E1", class_id="C2", team_id="TEAM_B")) + data = self.assertOk(api.pbl_domain_ref_list( + {"tenant_id": T1, "class_id": "C1", "page": 1, "page_size": 10})) + self.assertEqual(data["total"], 2) + self.assertEqual(len(data["items"]), 2) + data2 = self.assertOk(api.pbl_domain_ref_list( + {"tenant_id": T1, "ref_type": "entity"})) + self.assertEqual([i["ref_id"] for i in data2["items"]], ["E1"]) + data3 = self.assertOk(api.pbl_domain_ref_list( + {"tenant_id": T1, "page": 1, "page_size": 2})) + self.assertLessEqual(len(data3["items"]), 2, "page_size 必须生效") + + def test_list_requires_tenant(self): + self.assertErr(api.pbl_domain_ref_list({"ref_type": "world"}), + "PBL_DE_TENANT_MISSING") + + def test_list_tenant_isolation(self): + self.assertOk(self.bind("world", "W1", tenant_id=T1)) + data = self.assertOk(api.pbl_domain_ref_list({"tenant_id": T2})) + self.assertEqual(data["total"], 0, "T2 不得看到 T1 的关联记录") -class TestListRefs(BaseCase): - """接口 4：list_refs（过滤 + 分页 + 悬挂引用标记）""" +class TestCheckAccess(BaseCase): - def _seed(self): - run(api.bind_ref('world', 1, blueprint_id='BP-1', class_id='CLS-1', - team_id='TEAM_A', tenant_id=T1)) - run(api.bind_ref('world', 2, blueprint_id='BP-1', class_id='CLS-1', - team_id='TEAM_B', tenant_id=T1)) - run(api.bind_ref('scene', 11, blueprint_id='BP-2', class_id='CLS-1', - tenant_id=T1)) - run(api.bind_ref('entity', 101, class_id='CLS-2', tenant_id=T1)) - run(api.bind_ref('world', 1, class_id='CLS-9', tenant_id=T2)) + def test_check_access_bound_ok(self): + self.assertOk(self.bind("world", "W1", class_id="C1", team_id="TEAM_A")) + data = self.assertOk(api.pbl_domain_ref_check_access( + {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", + "class_id": "C1", "team_id": "TEAM_A"})) + self.assertTrue(data.get("allowed") or data.get("access"), + "已绑定且班级/团队匹配应放行：%r" % data) - def test_list_all_tenant_scoped(self): - self._seed() - res = run(api.list_refs({}, 1, 20, tenant_id=T1)) - self.assertEqual(res['total'], 4, 'T1 只应看到自己的 4 条关联') - self.assertTrue(all(i['tenant_id'] == T1 for i in res['items'])) + def test_check_access_other_team_denied(self): + self.assertOk(self.bind("world", "W1", class_id="C1", team_id="TEAM_A")) + resp = api.pbl_domain_ref_check_access( + {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "team_id": "TEAM_Z"}) + data = resp.get("data") or {} + if resp.get("success"): + self.assertFalse(data.get("allowed") and data.get("access", True), + "他团队不得放行：%r" % data) + else: + self.assertIn(resp.get("code"), ("PBL_DE_ACCESS_DENIED", "PBL_DE_NOT_FOUND")) - def test_list_filter_by_type_and_class(self): - self._seed() - res = run(api.list_refs({'ref_type': 'world'}, 1, 20, tenant_id=T1)) - self.assertEqual(res['total'], 2) - res2 = run(api.list_refs({'class_id': 'CLS-2'}, 1, 20, tenant_id=T1)) - self.assertEqual(res2['total'], 1) - self.assertEqual(res2['items'][0]['ref_type'], 'entity') - res3 = run(api.list_refs({'blueprint_id': 'BP-1'}, 1, 20, tenant_id=T1)) - self.assertEqual(res3['total'], 2) - res4 = run(api.list_refs({'team_id': 'TEAM_B'}, 1, 20, tenant_id=T1)) - self.assertEqual(res4['total'], 1) + def test_check_access_cross_tenant_denied(self): + self.assertOk(self.bind("world", "W1", tenant_id=T1)) + resp = api.pbl_domain_ref_check_access( + {"tenant_id": T2, "ref_type": "world", "ref_id": "W1"}) + data = resp.get("data") or {} + self.assertFalse(resp.get("success") and (data.get("allowed") or data.get("access")), + "跨租户不得放行") - def test_list_pagination(self): - self._seed() - page1 = run(api.list_refs({}, 1, 2, tenant_id=T1)) - page2 = run(api.list_refs({}, 2, 2, tenant_id=T1)) - self.assertEqual(len(page1['items']), 2) - self.assertEqual(len(page2['items']), 2) - self.assertEqual(page1['total'], 4) - ids1 = {i['id'] for i in page1['items']} - ids2 = {i['id'] for i in page2['items']} - self.assertFalse(ids1 & ids2, '分页结果不得重叠') - - def test_list_page_size_capped(self): - self._seed() - res = run(api.list_refs({}, 1, 9999, tenant_id=T1)) - self.assertEqual(res['size'], api.MAX_PAGE_SIZE) - - def test_list_invalid_ref_type_filter(self): - self.assertPblError(E_VALIDATION, - api.list_refs({'ref_type': 'bogus'}, 1, 20, tenant_id=T1)) - - def test_list_dangling_flag(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.conn.execute("DELETE FROM world WHERE id=1") # 复用模块删除基表记录 - self.conn.commit() - res = run(api.list_refs({'ref_type': 'world'}, 1, 20, tenant_id=T1)) - self.assertEqual(res['total'], 1) - self.assertTrue(res['items'][0]['dangling'], '基表已删的 ref 必须标记 dangling') - self.assertEqual(res['valid_total'], 0) + def test_check_access_requires_params(self): + self.assertErr(api.pbl_domain_ref_check_access({"tenant_id": T1}), + ("PBL_DE_PARAM_INVALID", "PBL_DE_REF_TYPE_INVALID"), + "缺 ref_type/ref_id") -class TestTenantIsolationQueries(BaseCase): - """接口 6~9：租户隔离查询封装（基表 + 扩展联合）""" +# ========================================================================== +# 2. 基础域只读契约（world / scene / entity） +# ========================================================================== +class TestBaseDomainReadOnly(BaseCase): - def test_list_worlds_by_tenant_only_bound(self): - run(api.bind_ref('world', 1, class_id='CLS-1', tenant_id=T1)) - worlds = run(api.list_worlds_by_tenant(T1)) - self.assertEqual(len(worlds), 1) - self.assertEqual(worlds[0]['id'], 1) - self.assertEqual(worlds[0]['name'], '世界A-火星基地') - self.assertEqual(worlds[0]['class_id'], 'CLS-1') - self.assertIsNotNone(worlds[0]['pbl_ref']) - # 未绑定的 world 2 对 T1 不可见 - self.assertNotIn(2, [w['id'] for w in worlds]) + def test_world_list_by_tenant(self): + data = self.assertOk(api.pbl_world_list_by_tenant({"tenant_id": T1})) + ids = sorted(str(i["id"]) for i in data["items"]) + self.assertEqual(ids, ["W1", "W2"], "只返回本租户世界") + self.assertTrue(data.get("readonly"), "必须标注只读") - def test_list_worlds_by_tenant_empty_when_no_binding(self): - self.assertEqual(run(api.list_worlds_by_tenant(T2)), []) + def test_world_list_tenant_isolation(self): + data = self.assertOk(api.pbl_world_list_by_tenant({"tenant_id": T2})) + self.assertEqual([str(i["id"]) for i in data["items"]], ["W3"]) - def test_list_worlds_by_class_filter(self): - run(api.bind_ref('world', 1, class_id='CLS-1', tenant_id=T1)) - run(api.bind_ref('world', 2, class_id='CLS-2', tenant_id=T1)) - got = run(api.list_worlds_by_tenant(T1, class_id='CLS-2')) - self.assertEqual([w['id'] for w in got], [2]) + def test_world_list_requires_tenant(self): + self.assertErr(api.pbl_world_list_by_tenant({}), "PBL_DE_TENANT_MISSING") - def test_list_worlds_filters_dangling(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.conn.execute("DELETE FROM world WHERE id=1") - self.conn.commit() - self.assertEqual(run(api.list_worlds_by_tenant(T1)), [], - '悬挂引用必须从联合结果中过滤') + def test_world_list_marks_bound(self): + self.assertOk(self.bind("world", "W1", blueprint_id="BP1")) + data = self.assertOk(api.pbl_world_list_by_tenant({"tenant_id": T1, "only_bound": 1})) + self.assertEqual([str(i["id"]) for i in data["items"]], ["W1"]) + self.assertTrue(data["items"][0]["pbl_bound"]) - def test_list_scenes_by_world_ok(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - run(api.bind_ref('scene', 11, tenant_id=T1)) - run(api.bind_ref('scene', 12, tenant_id=T1)) - scenes = run(api.list_scenes_by_world(1, T1)) - self.assertEqual(sorted(s['id'] for s in scenes), [11, 12]) - self.assertEqual(scenes[0]['world_id'], 1) + def test_world_get_context(self): + self.assertOk(self.bind("world", "W1", blueprint_id="BP1")) + data = self.assertOk(api.pbl_world_get_context( + {"tenant_id": T1, "world_id": "W1", "with_scenes": 1, "with_entities": 1})) + self.assertEqual(str(data["world"]["id"]), "W1") + self.assertEqual(sorted(str(s["id"]) for s in data.get("scenes", [])), ["S1", "S2"]) - def test_list_scenes_only_bound_visible(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - run(api.bind_ref('scene', 11, tenant_id=T1)) # 12 未绑定 - scenes = run(api.list_scenes_by_world(1, T1)) - self.assertEqual([s['id'] for s in scenes], [11]) + def test_world_get_context_missing(self): + self.assertErr(api.pbl_world_get_context({"tenant_id": T1, "world_id": "NOPE"}), + ("PBL_DE_NOT_FOUND", "PBL_DE_BASE_MISSING")) - def test_list_scenes_cross_tenant_forbidden(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.assertPblError(E_FORBIDDEN, api.list_scenes_by_world(1, T2)) + def test_world_get_context_cross_tenant(self): + self.assertErr(api.pbl_world_get_context({"tenant_id": T2, "world_id": "W1"}), + ("PBL_DE_NOT_FOUND", "PBL_DE_BASE_MISSING", "PBL_DE_ACCESS_DENIED"), + "W1 属 T1，T2 不得取其上下文") - def test_list_entities_by_scene_ok(self): - run(api.bind_ref('scene', 11, tenant_id=T1)) - run(api.bind_ref('entity', 101, tenant_id=T1)) - run(api.bind_ref('entity', 102, tenant_id=T1)) - ents = run(api.list_entities_by_scene(11, T1)) - self.assertEqual(sorted(e['id'] for e in ents), [101, 102]) - self.assertEqual(ents[0]['kind'], 'vehicle') + def test_scene_list_by_world(self): + data = self.assertOk(api.pbl_scene_list_by_world( + {"tenant_id": T1, "world_id": "W1"})) + self.assertEqual(sorted(str(i["id"]) for i in data["items"]), ["S1", "S2"]) - def test_list_entities_cross_tenant_forbidden(self): - run(api.bind_ref('scene', 11, tenant_id=T1)) - self.assertPblError(E_FORBIDDEN, api.list_entities_by_scene(11, T2)) + def test_scene_list_requires_world_id(self): + self.assertErr(api.pbl_scene_list_by_world({"tenant_id": T1}), + "PBL_DE_PARAM_INVALID") - def test_get_world_with_pbl_context_ok(self): - run(api.bind_ref('world', 1, blueprint_id='BP-1', class_id='CLS-1', - team_id='TEAM_A', ext_json={'mode': 'coop'}, tenant_id=T1)) - ctx = run(api.get_world_with_pbl_context(1, T1)) - self.assertEqual(ctx['name'], '世界A-火星基地') - self.assertEqual(ctx['pbl_context']['blueprint_id'], 'BP-1') - self.assertEqual(ctx['pbl_context']['class_id'], 'CLS-1') - self.assertEqual(ctx['pbl_context']['team_id'], 'TEAM_A') - self.assertEqual(ctx['pbl_context']['ext_json'], {'mode': 'coop'}) - self.assertEqual(ctx['pbl_context']['tenant_id'], T1) + def test_scene_list_cross_tenant_empty(self): + data = self.assertOk(api.pbl_scene_list_by_world( + {"tenant_id": T2, "world_id": "W1"})) + self.assertEqual(data["items"], [], "跨租户不得看到他租户场景") - def test_get_world_context_cross_tenant_forbidden(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.assertPblError(E_FORBIDDEN, api.get_world_with_pbl_context(1, T2)) + def test_entity_list_by_scene(self): + data = self.assertOk(api.pbl_entity_list_by_scene( + {"tenant_id": T1, "scene_id": "S1"})) + self.assertEqual(sorted(str(i["id"]) for i in data["items"]), ["E1", "E2"]) - def test_get_world_context_base_missing_not_found(self): - run(api.bind_ref('world', 1, tenant_id=T1)) - self.conn.execute("DELETE FROM world WHERE id=1") - self.conn.commit() - self.assertPblError(E_NOT_FOUND, api.get_world_with_pbl_context(1, T1)) + def test_entity_list_requires_scene_id(self): + self.assertErr(api.pbl_entity_list_by_scene({"tenant_id": T1}), + "PBL_DE_PARAM_INVALID") -class TestCheckRefAccess(BaseCase): - """接口 10：check_ref_access（租户+班级+团队三重匹配）""" +# ========================================================================== +# 3. 班级 / 团队维度契约 +# ========================================================================== +class TestTeamClass(BaseCase): - def test_access_granted(self): - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - self.assertTrue(run(api.check_ref_access('world', 1, T1))) - self.assertTrue(run(api.check_ref_access('world', 1, T1, - class_id='CLS-1', team_id='TEAM_A'))) + def test_team_bind_world_ok(self): + data = self.assertOk(api.pbl_team_bind_world( + {"tenant_id": T1, "world_id": "W1", "team_id": "TEAM_A", + "class_id": "C1", "operator_id": OP})) + self.assertEqual(data["team_id"], "TEAM_A") + self.assertEqual(data["ref_type"], "world") + self.assertEqual(len(self.rows()), 1) - def test_access_denied_cross_tenant(self): - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - self.assertFalse(run(api.check_ref_access('world', 1, T2))) + def test_team_bind_world_idempotent(self): + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W1", "team_id": "TEAM_A"}) + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W1", "team_id": "TEAM_A"}) + self.assertEqual(len(self.rows()), 1, "重复绑定不得产生重复行") - def test_access_denied_wrong_class_or_team(self): - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - self.assertFalse(run(api.check_ref_access('world', 1, T1, class_id='CLS-X'))) - self.assertFalse(run(api.check_ref_access('world', 1, T1, team_id='TEAM-X'))) + def test_team_bind_world_requires_team_id(self): + self.assertErr(api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W1"}), + "PBL_DE_PARAM_INVALID") - def test_access_denied_unbound(self): - self.assertFalse(run(api.check_ref_access('world', 2, T1))) + def test_team_bind_world_base_missing(self): + self.assertErr(api.pbl_team_bind_world( + {"tenant_id": T1, "world_id": "NOPE", "team_id": "TEAM_A"}), + ("PBL_DE_BASE_MISSING", "PBL_DE_NOT_FOUND")) - def test_access_denied_after_unbind(self): - run(api.bind_ref('scene', 11, tenant_id=T1)) - run(api.unbind_ref('scene', 11, tenant_id=T1)) - self.assertFalse(run(api.check_ref_access('scene', 11, T1))) + def test_team_bind_world_cross_tenant(self): + self.assertErr(api.pbl_team_bind_world( + {"tenant_id": T2, "world_id": "W1", "team_id": "TEAM_A"}), + ("PBL_DE_BASE_MISSING", "PBL_DE_NOT_FOUND", "PBL_DE_ACCESS_DENIED")) - def test_access_denied_invalid_input_no_raise(self): - self.assertFalse(run(api.check_ref_access('bogus', 1, T1))) - self.assertFalse(run(api.check_ref_access('world', 'x', T1))) - self.assertFalse(run(api.check_ref_access('world', 1, None))) + def test_team_world_list(self): + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W1", "team_id": "TEAM_A"}) + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W2", "team_id": "TEAM_B"}) + data = self.assertOk(api.pbl_team_world_list({"tenant_id": T1, "team_id": "TEAM_A"})) + items = data["items"] if isinstance(data, dict) else data + self.assertEqual(sorted(str(i.get("id") or i.get("world_id")) for i in items), ["W1"]) - def test_access_denied_dangling_base(self): - run(api.bind_ref('entity', 101, tenant_id=T1)) - self.conn.execute("DELETE FROM entity WHERE id=101") - self.conn.commit() - self.assertFalse(run(api.check_ref_access('entity', 101, T1))) - - -class TestTeamClassDimension(BaseCase): - """接口 11~13：团队/班级维度（US-13 共享世界支撑）""" - - def test_list_teams_by_class_groups(self): - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - run(api.bind_ref('scene', 11, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - run(api.bind_ref('world', 2, class_id='CLS-1', team_id='TEAM_B', tenant_id=T1)) - teams = run(api.list_teams_by_class('CLS-1', T1)) - self.assertEqual(len(teams), 2) - by_id = {t['team_id']: t for t in teams} - self.assertEqual(sorted(by_id['TEAM_A']['world_ids']), [1]) - self.assertEqual(sorted(by_id['TEAM_A']['scene_ids']), [11]) - self.assertEqual(by_id['TEAM_A']['entity_ids'], []) - self.assertEqual(len(by_id['TEAM_A']['ref_ids']), 2) - self.assertEqual(sorted(by_id['TEAM_B']['world_ids']), [2]) - - def test_list_teams_by_class_members_from_governance(self): - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - teams = run(api.list_teams_by_class('CLS-1', T1)) - members = teams[0]['members'] - self.assertEqual(len(members), 2, 'members 应只读取自 pbl_governance.pbl_team_member') - self.assertEqual({m['user_id'] for m in members}, {'stu01', 'stu02'}) - - def test_list_teams_by_class_tenant_scoped(self): - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - self.assertEqual(run(api.list_teams_by_class('CLS-1', T2)), []) - - def test_list_teams_by_class_requires_class_id(self): - self.assertPblError(E_VALIDATION, api.list_teams_by_class(None, T1)) - - def test_list_teams_without_governance_table_degrades(self): - """治理模块未部署（无 pbl_team_member 表）时降级为空成员列表，不报错。""" - fake_db.teardown() - self.conn.close() - self.conn, self.adapter = fake_db.setup(with_team_member=False) - run(api.bind_ref('world', 1, class_id='CLS-1', team_id='TEAM_A', tenant_id=T1)) - teams = run(api.list_teams_by_class('CLS-1', T1)) - self.assertEqual(len(teams), 1) - self.assertEqual(teams[0]['members'], []) - - def test_bind_team_to_world_creates_ref(self): - ref = run(api.bind_team_to_world(1, 'TEAM_A', T1, class_id='CLS-1', - blueprint_id='BP-1')) - self.assertEqual(ref['ref_type'], 'world') - self.assertEqual(ref['ref_id'], 1) - self.assertEqual(ref['team_id'], 'TEAM_A') - self.assertEqual(ref['class_id'], 'CLS-1') - - def test_bind_team_to_world_updates_existing(self): - run(api.bind_ref('world', 1, team_id='TEAM_A', tenant_id=T1)) - ref = run(api.bind_team_to_world(1, 'TEAM_B', T1)) - self.assertEqual(ref['team_id'], 'TEAM_B', '已有关联应幂等更新 team_id') - cnt = fake_db.scalar( - self.conn, - "SELECT COUNT(*) FROM pbl_domain_ref WHERE ref_type='world' AND ref_id=1") - self.assertEqual(cnt, 1, '不得产生重复关联行') - - def test_bind_team_to_world_duplicate_same_team(self): - run(api.bind_team_to_world(1, 'TEAM_A', T1)) - self.assertPblError(E_DUPLICATE, api.bind_team_to_world(1, 'TEAM_A', T1)) - - def test_bind_team_to_world_base_missing(self): - self.assertPblError(E_NOT_FOUND, api.bind_team_to_world(999, 'TEAM_A', T1)) - - def test_bind_team_to_world_requires_team_id(self): - self.assertPblError(E_VALIDATION, api.bind_team_to_world(1, None, T1)) - - def test_get_team_worlds(self): - run(api.bind_team_to_world(1, 'TEAM_A', T1)) - run(api.bind_team_to_world(2, 'TEAM_B', T1)) - worlds = run(api.get_team_worlds('TEAM_A', T1)) - self.assertEqual([w['id'] for w in worlds], [1]) - self.assertEqual(worlds[0]['team_id'], 'TEAM_A') - self.assertEqual(run(api.get_team_worlds('TEAM_A', T2)), [], - '跨租户不得返回他租户团队世界') - - def test_get_team_worlds_requires_team_id(self): - self.assertPblError(E_VALIDATION, api.get_team_worlds('', T1)) + def test_team_world_list_requires_team_id(self): + self.assertErr(api.pbl_team_world_list({"tenant_id": T1}), "PBL_DE_PARAM_INVALID") + + def test_team_world_list_cross_tenant_empty(self): + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W1", "team_id": "TEAM_A"}) + data = self.assertOk(api.pbl_team_world_list({"tenant_id": T2, "team_id": "TEAM_A"})) + items = data["items"] if isinstance(data, dict) else data + self.assertEqual(list(items), [], "跨租户不得返回他租户团队世界") + + def test_team_list_by_class(self): + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W1", + "team_id": "TEAM_A", "class_id": "C1"}) + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W2", + "team_id": "TEAM_B", "class_id": "C2"}) + data = self.assertOk(api.pbl_team_list_by_class({"tenant_id": T1, "class_id": "C1"})) + items = data["items"] if isinstance(data, dict) else data + self.assertEqual(len(items), 1, "只返回该班级团队") + + def test_team_list_by_class_requires_class_id(self): + self.assertErr(api.pbl_team_list_by_class({"tenant_id": T1}), "PBL_DE_PARAM_INVALID") +# ========================================================================== +# 4. 薄扩展铁律 / 契约齐备 / 设计对齐 +# ========================================================================== class TestThinExtensionInvariants(BaseCase): - """薄扩展铁律：不改基表 + 表总账 + 契约齐备""" - def test_base_tables_untouched(self): - before = {t: fake_db.scalar(self.conn, "SELECT COUNT(*) FROM %s" % t) - for t in ('world', 'scene', 'entity')} - run(api.bind_ref('world', 1, class_id='C', team_id='T', tenant_id=T1)) - run(api.bind_ref('scene', 11, tenant_id=T1)) - run(api.bind_ref('entity', 101, tenant_id=T1)) - run(api.update_ref('world', 1, {'class_id': 'C2'}, tenant_id=T1)) - run(api.bind_team_to_world(2, 'TEAM_B', T1)) - # 读路径（联合查询）——必须在解绑前跑，解绑后 scene 11 对本租户即 403 - run(api.list_worlds_by_tenant(T1)) - run(api.list_scenes_by_world(1, T1)) - run(api.list_entities_by_scene(11, T1)) - run(api.get_world_with_pbl_context(1, T1)) - run(api.check_ref_access('world', 1, T1)) - run(api.list_teams_by_class('C', T1)) - run(api.get_team_worlds('T', T1)) - run(api.list_refs({}, 1, 20, tenant_id=T1)) - # 写路径收尾：解绑（软删，只动自有表） - run(api.unbind_ref('scene', 11, tenant_id=T1)) - run(api.unbind_ref('entity', 101, tenant_id=T1)) - after = {t: fake_db.scalar(self.conn, "SELECT COUNT(*) FROM %s" % t) - for t in ('world', 'scene', 'entity')} - self.assertEqual(before, after, '基表行数不得变化（零写入）') - # 基表结构零 ALTER：列集合不变 - for table in ('world', 'scene', 'entity'): - cols = [r[1] for r in fake_db.raw_sql( - self.conn, "PRAGMA table_info(%s)" % table)] - self.assertNotIn('tenant_id', cols, '禁止给基表加 tenant_id 列') + def test_base_tables_never_written(self): + before = {t: len(self.sor.tables.get(t, [])) for t in ("world", "scene", "entity")} + self.assertOk(self.bind("world", "W1", class_id="C1", team_id="T")) + self.assertOk(self.bind("scene", "S1")) + self.assertOk(self.bind("entity", "E1")) + api.pbl_domain_ref_update({"tenant_id": T1, "ref_type": "world", + "ref_id": "W1", "class_id": "C2"}) + api.pbl_team_bind_world({"tenant_id": T1, "world_id": "W2", "team_id": "TB"}) + api.pbl_domain_ref_unbind({"tenant_id": T1, "ref_type": "scene", "ref_id": "S1"}) + self.assertEqual(self.sor.base_write_attempts, [], + "基表 world/scene/entity 必须零写入（Q-OPEN-3 薄扩展铁律）") + after = {t: len(self.sor.tables.get(t, [])) for t in ("world", "scene", "entity")} + self.assertEqual(before, after, "基表行数不得变化") - def test_own_tables_single_and_matches_models(self): - self.assertEqual(init_mod.OWN_TABLES, ['pbl_domain_ref']) - models_dir = os.path.join(ROOT, 'models') - files = sorted(f for f in os.listdir(models_dir) if f.endswith('.json')) - self.assertEqual(files, ['pbl_domain_ref.json'], - 'models/ 必须与 OWN_TABLES 一一对应（QC #4）') + def test_only_one_new_table(self): + self.assertOk(self.bind("world", "W1")) + new_tables = set(self.sor.tables) - set(fake_db.BASE_FIXTURES) + self.assertEqual(new_tables, {TABLE}, + "M8 只允许新增 1 张关联表 pbl_domain_ref，不得新增 world/scene/entity 三张 PBL 表") - def test_model_json_four_sections(self): - path = os.path.join(ROOT, 'models', 'pbl_domain_ref.json') - model = json.load(open(path, encoding='utf-8')) - for key in ('summary', 'fields', 'indexes', 'codes'): - self.assertIn(key, model, '表定义四段式缺 %s' % key) - self.assertIsInstance(model['summary'], list, 'summary 必须是数组（QC #3）') - self.assertTrue(all(isinstance(s, str) for s in model['summary'])) - self.assertEqual(model['summary'][0], 'pbl_domain_ref') - primaries = [f for f in model['fields'] if f.get('primary')] - self.assertEqual(len(primaries), 1) - self.assertEqual(primaries[0]['name'], 'id') - uniq = [i for i in model['indexes'] if i.get('unique') and not i.get('primary')] - self.assertIn(['tenant_id', 'ref_type', 'ref_id'], - [i['fields'] for i in uniq], - '必须有 UNIQUE(tenant_id,ref_type,ref_id)（设计 §2）') + def test_ext_field_name_matches_design(self): + self.assertEqual(EXT_FIELD, "ext_json", + "扩展列名必须为设计 §J1 权威名 ext_json（非 ext）") + self.assertOk(self.bind("world", "W1", ext_json='{"k":"v"}')) + row = self.rows()[0] + self.assertIn("ext_json", row, "落库列名必须是 ext_json") + self.assertNotIn("ext", row, "不得残留旧列名 ext") - def test_contract_13_interfaces_all_callable(self): - self.assertEqual(len(api.CONTRACT_INTERFACES), 13) - for name in api.CONTRACT_INTERFACES: - self.assertTrue(callable(getattr(api, name, None)), '缺实现：%s' % name) + def test_ref_types_cover_three_domains(self): + self.assertEqual(tuple(REF_TYPES), ("world", "scene", "entity")) - def test_contract_map_covers_dspy_files(self): - cmap = init_mod.get_contract_map() - self.assertEqual(len(cmap), 13) - api_dir = os.path.join(ROOT, 'wwwroot', 'api') - existing = set(os.listdir(api_dir)) - for name, (impl, dspy) in cmap.items(): - self.assertTrue(callable(getattr(api, name, None)), '缺实现：%s' % name) - self.assertIn(os.path.basename(dspy), existing, - '契约 %s 缺 dspy 端点 %s（QC #5）' % (name, dspy)) - self.assertTrue(impl.startswith('pbl_domain_ext/api.py:')) + def test_all_13_contracts_callable_and_registered(self): + from pbl_domain_ext import init as mod_init + names = [n for n in mod_init.CONTRACT_FUNCS if n != "pbl_domain_materialize_game_definition"] + self.assertEqual(len(names), 13, "契约函数应为 13 个") + missing_impl = [n for n in names if not callable(getattr(api, n, None))] + self.assertEqual(missing_impl, [], "契约函数必须在 api.py 实现") + info = mod_init._register_serverenv() + self.assertEqual(info["missing"], [], "全部契约必须注册到 ServerEnv（dspy 直接按名调用）") + for n in names: + self.assertIn(n, info["registered"]) - def test_load_path_registers_all_dspy(self): - sys.path.insert(0, os.path.join(ROOT, 'scripts')) - import load_path as lp - self.assertEqual(lp.selfcheck(), [], 'load_path 自检必须无问题（QC #6）') - registered = set(lp.API_PATHS) | set(lp.UI_PATHS) - api_dir = os.path.join(ROOT, 'wwwroot', 'api') - for fname in os.listdir(api_dir): - if fname.endswith('.dspy'): - self.assertIn('/pbl_domain_ext/api/%s' % fname, registered, - '端点 %s 未在 load_path.py 注册 → 上线 403' % fname) - self.assertIn('/pbl_domain_ext/index.ui', registered) - self.assertEqual(len(lp.API_PATHS), 13) - for path in lp.ALL_PATHS: - self.assertNotIn('*', path, '禁止通配符注册') + def test_dspy_files_match_contracts(self): + from pbl_domain_ext import init as mod_init + api_dir = os.path.join(ROOT, "wwwroot", "api") + for name, _action, _w in mod_init.API_ROUTES: + path = os.path.join(api_dir, name + ".dspy") + self.assertTrue(os.path.isfile(path), "缺少 dspy 契约文件 %s" % path) + body = open(path, encoding="utf-8").read() + self.assertIn(name + "(", body, "%s.dspy 必须调用同名契约函数" % name) + self.assertNotIn("import ", body, "dspy 禁止 import（全局已预载）") + self.assertIn("return", body, "dspy 必须显式 return") - def test_load_module_registers_env_functions(self): - env = init_mod.load_pbl_domain_ext() - for name in api.CONTRACT_INTERFACES: - self.assertTrue(callable(getattr(env, 'pbl_%s' % name, None)), - 'env.pbl_%s 未注册' % name) - self.assertTrue(callable(getattr(env, name, None)), - 'env.%s（设计原名）未注册' % name) - info = env.pbl_domain_ext_module_info - self.assertEqual(info['own_tables'], ['pbl_domain_ref']) - self.assertFalse(info['base_table_altered']) - self.assertEqual(len(info['contracts']), 13) + def test_dispatch_routes_all_contracts(self): + for name in ("pbl_domain_ref_bind", "pbl_world_list_by_tenant", + "pbl_team_bind_world", "pbl_scene_list_by_world"): + resp = api.dispatch(name, {"tenant_id": T1, "ref_type": "world", + "ref_id": "W1", "world_id": "W1", + "scene_id": "S1", "team_id": "TEAM_A"}) + self.assertIsInstance(resp, dict, "dispatch(%s) 必须返回 dict" % name) - def test_package_exports_match_init(self): + def test_materialize_game_definition_exported(self): + """历史 QC 硬门禁：pbl_compiler 曾引用该符号，必须真实存在且可调用。""" import pbl_domain_ext as pkg - for name in api.CONTRACT_INTERFACES: - self.assertTrue(callable(getattr(pkg, name, None)), - '__init__.py 未导出 %s（三处同步注册之 ②）' % name) - self.assertTrue(callable(pkg.load_pbl_domain_ext)) - - def test_base_layer_readonly_projection(self): - self.assertTrue(run(base_mod.base_exists('world', 1))) - self.assertFalse(run(base_mod.base_exists('world', 999))) - rows = run(base_mod.fetch_children('scene', 1)) - self.assertEqual(sorted(r['id'] for r in rows), [11, 12]) - col = run(base_mod.resolve_column('scene', base_mod.SCENE_PARENT_CANDIDATES)) - self.assertEqual(col, 'world_id') + self.assertTrue(callable(getattr(pkg, "pbl_domain_materialize_game_definition", None))) + self.assertTrue(callable(getattr(api, "pbl_domain_materialize_game_definition", None))) + resp = api.pbl_domain_materialize_game_definition({"tenant_id": T1, "world_id": "W1"}) + self.assertIsInstance(resp, dict) -if __name__ == '__main__': +if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/wwwroot/api/pbl_domain_ref_bind.dspy b/wwwroot/api/pbl_domain_ref_bind.dspy index 0926f56..4c68bec 100644 --- a/wwwroot/api/pbl_domain_ref_bind.dspy +++ b/wwwroot/api/pbl_domain_ref_bind.dspy @@ -1,29 +1,19 @@ -# api/pbl_domain_ref_bind.dspy — 契约端点：bind_ref（设计 §3.1 接口 1） -# 实现：pbl_domain_ext/api.py:bind_ref（经 init.py 注册为 env.pbl_bind_ref） -# 入参：ref_type(world/scene/entity) ref_id blueprint_id class_id team_id ext_json tenant_id -# 说明：ext_json 原样透传（dict 或 JSON 字符串均可，由 api._dump_ext 统一校验/序列化） -debug('pbl_domain_ref_bind.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_domain_ref_bind.dspy — 契约端点：绑定基础域对象到 PBL 蓝图/班级/团队（幂等 upsert） +# 实现：pbl_domain_ext/api.py:pbl_domain_ref_bind（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_domain_ref_bind） +# 入参：tenant_id* ref_type*(world|scene|entity) ref_id* ref_code ref_name blueprint_id class_id team_id ext_json operator_id +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_domain_ref_bind.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_bind_ref( - params_kw.get('ref_type'), - params_kw.get('ref_id'), - blueprint_id=params_kw.get('blueprint_id'), - class_id=params_kw.get('class_id'), - team_id=params_kw.get('team_id'), - ext_json=params_kw.get('ext_json'), - tenant_id=params_kw.get('tenant_id'), - ) - debug('pbl_domain_ref_bind.dspy: OK ref_type=%s ref_id=%s' - % (params_kw.get('ref_type'), params_kw.get('ref_id'))) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_domain_ref_bind(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_domain_ref_bind.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_domain_ref_bind.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_domain_ref_bind.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_domain_ref_check_access.dspy b/wwwroot/api/pbl_domain_ref_check_access.dspy index bfecebb..333617d 100644 --- a/wwwroot/api/pbl_domain_ref_check_access.dspy +++ b/wwwroot/api/pbl_domain_ref_check_access.dspy @@ -1,24 +1,19 @@ -# api/pbl_domain_ref_check_access.dspy — 契约端点：check_ref_access（设计 §3.2 接口 10） -# 实现：pbl_domain_ext/api.py:check_ref_access（env.pbl_check_ref_access） -# 出参：bool（租户+班级+团队三重匹配；不抛异常，供运行时热路径判定 F-RT-02） -debug('pbl_domain_ref_check_access.dspy: START params_kw=%s' % dict(params_kw)) +# api/pbl_domain_ref_check_access.dspy — 契约端点：访问判定：当前租户/班级/团队能否访问 ref_type+ref_id +# 实现：pbl_domain_ext/api.py:pbl_domain_ref_check_access（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_domain_ref_check_access） +# 入参：tenant_id* ref_type* ref_id* class_id team_id +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_domain_ref_check_access.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _allowed = await pbl_check_ref_access( - params_kw.get('ref_type'), - params_kw.get('ref_id'), - params_kw.get('tenant_id'), - class_id=params_kw.get('class_id'), - team_id=params_kw.get('team_id'), - ) - debug('pbl_domain_ref_check_access.dspy: OK allowed=%s' % _allowed) - return {'success': True, 'error_code': None, 'message': 'ok', - 'data': {'allowed': bool(_allowed)}, - 'allowed': bool(_allowed), - 'http_status': 200 if _allowed else 403} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_domain_ref_check_access(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_domain_ref_check_access.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - debug('pbl_domain_ref_check_access.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'data': {'allowed': False}, 'allowed': False, 'http_status': 500} + debug('pbl_domain_ref_check_access.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_domain_ref_get.dspy b/wwwroot/api/pbl_domain_ref_get.dspy index f6a260b..a8a7f74 100644 --- a/wwwroot/api/pbl_domain_ref_get.dspy +++ b/wwwroot/api/pbl_domain_ref_get.dspy @@ -1,22 +1,19 @@ -# api/pbl_domain_ref_get.dspy — 契约端点：get_ref（设计 §3.1 接口 3） -# 实现：pbl_domain_ext/api.py:get_ref（env.pbl_get_ref） -debug('pbl_domain_ref_get.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_domain_ref_get.dspy — 契约端点：取单条关联记录（with_base=1 时附基表只读视图） +# 实现：pbl_domain_ext/api.py:pbl_domain_ref_get（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_domain_ref_get） +# 入参：tenant_id* id 或 ref_type+ref_id with_base +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_domain_ref_get.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_get_ref( - params_kw.get('ref_type'), - params_kw.get('ref_id'), - tenant_id=params_kw.get('tenant_id'), - ) - debug('pbl_domain_ref_get.dspy: OK id=%s' % (_data or {}).get('id')) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_domain_ref_get(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_domain_ref_get.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_domain_ref_get.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_domain_ref_get.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_domain_ref_list.dspy b/wwwroot/api/pbl_domain_ref_list.dspy index 483b2e4..b16f2b8 100644 --- a/wwwroot/api/pbl_domain_ref_list.dspy +++ b/wwwroot/api/pbl_domain_ref_list.dspy @@ -1,37 +1,19 @@ -# api/pbl_domain_ref_list.dspy — 契约端点：list_refs（设计 §3.1 接口 4） -# 实现：pbl_domain_ext/api.py:list_refs（env.pbl_list_refs） -# 入参：filters{ref_type,blueprint_id,class_id,team_id,ref_id} page size tenant_id -debug('pbl_domain_ref_list.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} - -# filters 兼容两种传法：整体 JSON 字符串 / 平铺字段（DSPY 必须转发全部客户端参数，禁硬编码） -_filters = params_kw.get('filters') -if isinstance(_filters, str) and _filters.strip(): - try: - _filters = json.loads(_filters) - except Exception: - _filters = None -if not isinstance(_filters, dict): - _filters = {} -for _k in ('ref_type', 'blueprint_id', 'class_id', 'team_id', 'ref_id'): - if _k not in _filters and params_kw.get(_k) not in (None, ''): - _filters[_k] = params_kw.get(_k) +# api/pbl_domain_ref_list.dspy — 契约端点：分页查询关联记录 +# 实现：pbl_domain_ext/api.py:pbl_domain_ref_list（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_domain_ref_list） +# 入参：tenant_id* ref_type bind_state blueprint_id team_id class_id ref_ids keyword page page_size +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_domain_ref_list.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_list_refs( - _filters, - params_kw.get('page', 1), - params_kw.get('size', 20), - tenant_id=params_kw.get('tenant_id'), - ) - debug('pbl_domain_ref_list.dspy: OK total=%s' % (_data or {}).get('total')) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_domain_ref_list(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_domain_ref_list.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_domain_ref_list.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_domain_ref_list.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_domain_ref_unbind.dspy b/wwwroot/api/pbl_domain_ref_unbind.dspy index 01ebadc..e724858 100644 --- a/wwwroot/api/pbl_domain_ref_unbind.dspy +++ b/wwwroot/api/pbl_domain_ref_unbind.dspy @@ -1,22 +1,19 @@ -# api/pbl_domain_ref_unbind.dspy — 契约端点：unbind_ref（设计 §3.1 接口 2） -# 实现：pbl_domain_ext/api.py:unbind_ref（env.pbl_unbind_ref）；软删 is_deleted=1 -debug('pbl_domain_ref_unbind.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_domain_ref_unbind.dspy — 契约端点：解绑（软删 is_deleted=1，不级联删除基表 world/scene/entity） +# 实现：pbl_domain_ext/api.py:pbl_domain_ref_unbind（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_domain_ref_unbind） +# 入参：tenant_id* id 或 ref_type+ref_id operator_id +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_domain_ref_unbind.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_unbind_ref( - params_kw.get('ref_type'), - params_kw.get('ref_id'), - tenant_id=params_kw.get('tenant_id'), - ) - debug('pbl_domain_ref_unbind.dspy: OK result=%s' % _data) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_domain_ref_unbind(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_domain_ref_unbind.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_domain_ref_unbind.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_domain_ref_unbind.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_domain_ref_update.dspy b/wwwroot/api/pbl_domain_ref_update.dspy index 31829f9..4918174 100644 --- a/wwwroot/api/pbl_domain_ref_update.dspy +++ b/wwwroot/api/pbl_domain_ref_update.dspy @@ -1,36 +1,19 @@ -# api/pbl_domain_ref_update.dspy — 契约端点：update_ref（设计 §3.1 接口 5） -# 实现：pbl_domain_ext/api.py:update_ref（env.pbl_update_ref） -# 入参：ref_type ref_id data{class_id,team_id,ext_json,blueprint_id} tenant_id -debug('pbl_domain_ref_update.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} - -_data_in = params_kw.get('data') -if isinstance(_data_in, str) and _data_in.strip(): - try: - _data_in = json.loads(_data_in) - except Exception: - _data_in = None -if not isinstance(_data_in, dict): - _data_in = {} -for _k in ('blueprint_id', 'class_id', 'team_id', 'ext_json'): - if _k not in _data_in and params_kw.get(_k) is not None: - _data_in[_k] = params_kw.get(_k) +# api/pbl_domain_ref_update.dspy — 契约端点：更新关联记录（ext_json / blueprint_id / class_id / team_id） +# 实现：pbl_domain_ext/api.py:pbl_domain_ref_update（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_domain_ref_update） +# 入参：tenant_id* id 或 ref_type+ref_id ext_json blueprint_id class_id team_id operator_id +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_domain_ref_update.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_update_ref( - params_kw.get('ref_type'), - params_kw.get('ref_id'), - _data_in, - tenant_id=params_kw.get('tenant_id'), - ) - debug('pbl_domain_ref_update.dspy: OK id=%s' % (_data or {}).get('id')) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_domain_ref_update(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_domain_ref_update.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_domain_ref_update.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_domain_ref_update.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_entity_list_by_scene.dspy b/wwwroot/api/pbl_entity_list_by_scene.dspy index 0c79679..f8d1c97 100644 --- a/wwwroot/api/pbl_entity_list_by_scene.dspy +++ b/wwwroot/api/pbl_entity_list_by_scene.dspy @@ -1,23 +1,19 @@ -# api/pbl_entity_list_by_scene.dspy — 契约端点：list_entities_by_scene（设计 §3.2 接口 8） -# 实现：pbl_domain_ext/api.py:list_entities_by_scene（env.pbl_list_entities_by_scene） -# 越权：scene 非本租户绑定 → PBL_E_FORBIDDEN(403)（US-21） -debug('pbl_entity_list_by_scene.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_entity_list_by_scene.dspy — 契约端点：只读：按 scene_id 列实体（基表 entity + 关联表过滤） +# 实现：pbl_domain_ext/api.py:pbl_entity_list_by_scene（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_entity_list_by_scene） +# 入参：tenant_id* scene_id* keyword page page_size +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_entity_list_by_scene.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_list_entities_by_scene( - params_kw.get('scene_id'), - params_kw.get('tenant_id'), - ) - debug('pbl_entity_list_by_scene.dspy: OK count=%s' % len(_data or [])) - return {'success': True, 'error_code': None, 'message': 'ok', - 'data': {'items': _data, 'total': len(_data or [])}} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_entity_list_by_scene(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_entity_list_by_scene.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_entity_list_by_scene.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_entity_list_by_scene.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_scene_list_by_world.dspy b/wwwroot/api/pbl_scene_list_by_world.dspy index 1af8083..eb7d4fc 100644 --- a/wwwroot/api/pbl_scene_list_by_world.dspy +++ b/wwwroot/api/pbl_scene_list_by_world.dspy @@ -1,23 +1,19 @@ -# api/pbl_scene_list_by_world.dspy — 契约端点：list_scenes_by_world（设计 §3.2 接口 7） -# 实现：pbl_domain_ext/api.py:list_scenes_by_world（env.pbl_list_scenes_by_world） -# 越权：world 非本租户绑定 → PBL_E_FORBIDDEN(403)（US-21） -debug('pbl_scene_list_by_world.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_scene_list_by_world.dspy — 契约端点：只读：按 world_id 列场景（基表 scene + 关联表过滤） +# 实现：pbl_domain_ext/api.py:pbl_scene_list_by_world（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_scene_list_by_world） +# 入参：tenant_id* world_id* keyword page page_size +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_scene_list_by_world.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_list_scenes_by_world( - params_kw.get('world_id'), - params_kw.get('tenant_id'), - ) - debug('pbl_scene_list_by_world.dspy: OK count=%s' % len(_data or [])) - return {'success': True, 'error_code': None, 'message': 'ok', - 'data': {'items': _data, 'total': len(_data or [])}} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_scene_list_by_world(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_scene_list_by_world.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_scene_list_by_world.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_scene_list_by_world.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_team_bind_world.dspy b/wwwroot/api/pbl_team_bind_world.dspy index 93b1c3a..e8de6b2 100644 --- a/wwwroot/api/pbl_team_bind_world.dspy +++ b/wwwroot/api/pbl_team_bind_world.dspy @@ -1,34 +1,19 @@ -# api/pbl_team_bind_world.dspy — 契约端点：bind_team_to_world（设计 §3.3 接口 12） -# 实现：pbl_domain_ext/api.py:bind_team_to_world（env.pbl_bind_team_to_world） -# 错误分支：PBL_E_NOT_FOUND（world 基表不存在）/ PBL_E_DUPLICATE（同团队重复绑定） -debug('pbl_team_bind_world.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} - -_ext = params_kw.get('ext_json') -if isinstance(_ext, str) and _ext.strip(): - try: - _ext = json.loads(_ext) - except Exception: - pass +# api/pbl_team_bind_world.dspy — 契约端点：团队绑定世界（写 pbl_domain_ref ref_type=world team_id=...，幂等） +# 实现：pbl_domain_ext/api.py:pbl_team_bind_world（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_team_bind_world） +# 入参：tenant_id* world_id* team_id* class_id operator_id +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_team_bind_world.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_bind_team_to_world( - params_kw.get('world_id'), - params_kw.get('team_id'), - params_kw.get('tenant_id'), - class_id=params_kw.get('class_id'), - blueprint_id=params_kw.get('blueprint_id'), - ext_json=_ext, - ) - debug('pbl_team_bind_world.dspy: OK world_id=%s team_id=%s' - % (params_kw.get('world_id'), params_kw.get('team_id'))) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_team_bind_world(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_team_bind_world.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_team_bind_world.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_team_bind_world.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_team_list_by_class.dspy b/wwwroot/api/pbl_team_list_by_class.dspy index 330364a..efc974d 100644 --- a/wwwroot/api/pbl_team_list_by_class.dspy +++ b/wwwroot/api/pbl_team_list_by_class.dspy @@ -1,23 +1,19 @@ -# api/pbl_team_list_by_class.dspy — 契约端点：list_teams_by_class（设计 §3.3 接口 11） -# 实现：pbl_domain_ext/api.py:list_teams_by_class（env.pbl_list_teams_by_class） -# 出参：list[{team_id,members[],ref_ids[],world_ids[],scene_ids[],entity_ids[]}]（US-13 共享世界分组） -debug('pbl_team_list_by_class.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_team_list_by_class.dspy — 契约端点：只读：班级下团队及其绑定世界 +# 实现：pbl_domain_ext/api.py:pbl_team_list_by_class（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_team_list_by_class） +# 入参：tenant_id* class_id* keyword page page_size +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_team_list_by_class.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_list_teams_by_class( - params_kw.get('class_id'), - params_kw.get('tenant_id'), - ) - debug('pbl_team_list_by_class.dspy: OK teams=%s' % len(_data or [])) - return {'success': True, 'error_code': None, 'message': 'ok', - 'data': {'items': _data, 'total': len(_data or [])}} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_team_list_by_class(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_team_list_by_class.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_team_list_by_class.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_team_list_by_class.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_team_world_list.dspy b/wwwroot/api/pbl_team_world_list.dspy index dfd8b83..6f0ab59 100644 --- a/wwwroot/api/pbl_team_world_list.dspy +++ b/wwwroot/api/pbl_team_world_list.dspy @@ -1,23 +1,19 @@ -# api/pbl_team_world_list.dspy — 契约端点：get_team_worlds（设计 §3.3 接口 13） -# 实现：pbl_domain_ext/api.py:get_team_worlds（env.pbl_get_team_worlds） -# 出参：list[world+ref]（仅本租户 + 本团队绑定的 world） -debug('pbl_team_world_list.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_team_world_list.dspy — 契约端点：只读：团队可访问的世界清单 +# 实现：pbl_domain_ext/api.py:pbl_team_world_list（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_team_world_list） +# 入参：tenant_id* team_id* class_id page page_size +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_team_world_list.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_get_team_worlds( - params_kw.get('team_id'), - params_kw.get('tenant_id'), - ) - debug('pbl_team_world_list.dspy: OK count=%s' % len(_data or [])) - return {'success': True, 'error_code': None, 'message': 'ok', - 'data': {'items': _data, 'total': len(_data or [])}} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_team_world_list(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_team_world_list.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_team_world_list.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_team_world_list.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_world_get_context.dspy b/wwwroot/api/pbl_world_get_context.dspy index 438beaa..67b91ef 100644 --- a/wwwroot/api/pbl_world_get_context.dspy +++ b/wwwroot/api/pbl_world_get_context.dspy @@ -1,22 +1,19 @@ -# api/pbl_world_get_context.dspy — 契约端点：get_world_with_pbl_context（设计 §3.2 接口 9） -# 实现：pbl_domain_ext/api.py:get_world_with_pbl_context（env.pbl_get_world_with_pbl_context） -# 出参：world 基表字段 + pbl_context{tenant_id,blueprint_id,class_id,team_id,ext_json} -debug('pbl_world_get_context.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} +# api/pbl_world_get_context.dspy — 契约端点：只读：世界上下文聚合（world + 其下 scene/entity 关联） +# 实现：pbl_domain_ext/api.py:pbl_world_get_context（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_world_get_context） +# 入参：tenant_id* world_id* with_scenes with_entities +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_world_get_context.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_get_world_with_pbl_context( - params_kw.get('world_id'), - params_kw.get('tenant_id'), - ) - debug('pbl_world_get_context.dspy: OK world_id=%s' % params_kw.get('world_id')) - return {'success': True, 'error_code': None, 'message': 'ok', 'data': _data} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_world_get_context(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_world_get_context.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_world_get_context.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_world_get_context.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/api/pbl_world_list_by_tenant.dspy b/wwwroot/api/pbl_world_list_by_tenant.dspy index 3791df6..90a4582 100644 --- a/wwwroot/api/pbl_world_list_by_tenant.dspy +++ b/wwwroot/api/pbl_world_list_by_tenant.dspy @@ -1,36 +1,19 @@ -# api/pbl_world_list_by_tenant.dspy — 契约端点：list_worlds_by_tenant（设计 §3.2 接口 6） -# 实现：pbl_domain_ext/api.py:list_worlds_by_tenant（env.pbl_list_worlds_by_tenant） -# 隔离逻辑：基表 world 无租户列 → 先查 pbl_domain_ref 白名单再只读回查基表 -debug('pbl_world_list_by_tenant.dspy: START params_kw=%s' % dict(params_kw)) - -_HTTP = {'PBL_E_VALIDATION': 400, 'PBL_E_NOT_FOUND': 404, 'PBL_E_DUPLICATE': 409, - 'PBL_E_FORBIDDEN': 403, 'PBL_E_INTERNAL': 500} - -_filters = params_kw.get('filters') -if isinstance(_filters, str) and _filters.strip(): - try: - _filters = json.loads(_filters) - except Exception: - _filters = None -if not isinstance(_filters, dict): - _filters = {} -for _k in ('blueprint_id', 'team_id'): - if _k not in _filters and params_kw.get(_k) not in (None, ''): - _filters[_k] = params_kw.get(_k) +# api/pbl_world_list_by_tenant.dspy — 契约端点：只读：本租户可见世界（基表 world + 关联表过滤） +# 实现：pbl_domain_ext/api.py:pbl_world_list_by_tenant（init.py 经 load_pbl_domain_ext() 注册为 ServerEnv 全局 env.pbl_world_list_by_tenant） +# 入参：tenant_id* keyword status page page_size +# 规范：module-development-spec §2.5 —— dspy 无 import、显式 return、转发**全部**客户端参数（禁硬编码 dispatch 字段） +# 薄扩展铁律：只读写关联表 pbl_domain_ref，基表 world/scene/entity 一律只读（Q-OPEN-3） +debug('pbl_world_list_by_tenant.dspy: START params_kw=%s' % (dict(params_kw) if params_kw else {})) try: - _data = await pbl_list_worlds_by_tenant( - params_kw.get('tenant_id'), - class_id=params_kw.get('class_id'), - filters=_filters, - ) - debug('pbl_world_list_by_tenant.dspy: OK count=%s' % len(_data or [])) - return {'success': True, 'error_code': None, 'message': 'ok', - 'data': {'items': _data, 'total': len(_data or [])}} + _params = dict(params_kw) if params_kw else {} + _resp = pbl_world_list_by_tenant(_params) + if not isinstance(_resp, dict): + _resp = {'success': False, 'code': 'PBL_DE_INTERNAL', + 'message': '契约函数返回非 dict', 'data': None} + debug('pbl_world_list_by_tenant.dspy: DONE success=%s code=%s' % (_resp.get('success'), _resp.get('code'))) + return _resp except Exception as exc: - _code = getattr(exc, 'code', None) or 'PBL_E_INTERNAL' - _msg = getattr(exc, 'message', None) or str(exc) - _detail = getattr(exc, 'detail', None) or {} - debug('pbl_world_list_by_tenant.dspy: FAIL code=%s msg=%s' % (_code, _msg)) - return {'success': False, 'error_code': _code, 'message': _msg, - 'detail': _detail, 'http_status': _HTTP.get(_code, 500)} + debug('pbl_world_list_by_tenant.dspy: FAIL %s' % format_exc()) + return {'success': False, 'code': 'PBL_DE_INTERNAL', 'message': str(exc), + 'detail': format_exc(), 'http_status': 500, 'data': None} diff --git a/wwwroot/index.ui b/wwwroot/index.ui index 8042dbc..950ef13 100644 --- a/wwwroot/index.ui +++ b/wwwroot/index.ui @@ -1,129 +1,588 @@ { - "type": "panel", - "id": "pbl_domain_ext_index", - "title": "PBL 基础域薄扩展 world/scene/entity（M8）", - "layout": "vbox", - "items": [ + "widgettype": "VBox", + "options": { + "width": "100%", + "height": "100%", + "padding": "20px", + "backgroundColor": "#F5F7FA" + }, + "subwidgets": [ { - "type": "html", - "id": "hdr", - "html": "

PBL 基础域薄扩展契约面板

world / scene / entity 三模块的 PBL 侧薄扩展：仅新增关联表 pbl_domain_ref 与只读查询契约，不改三张基表结构（Q-OPEN-3）。下列 13 个契约卡片 url 全部使用 entire_url() 生成，避免 RBAC 403 与路由错误。

" + "widgettype": "Text", + "options": { + "label": "PBL 基础域薄扩展 (M8) — world / scene / entity 契约导航", + "fontSize": "22px", + "fontWeight": "bold" + } }, { - "type": "panel", - "id": "grp_ref", - "title": "一、关联表 pbl_domain_ref 契约（6）", - "layout": "hbox", - "items": [ + "widgettype": "Text", + "options": { + "label": "薄扩展边界：仅新增关联表 pbl_domain_ref，不修改 world / scene / entity 三张基表结构 (Q-OPEN-3)。下列 13 个契约入口的 url 一律使用 entire_url 包裹的 /pbl_domain_ext/api/xxx.dspy 绝对路径（module-development-spec 2.1 / 2.3c），避免 RBAC 403 与路由错误。", + "fontSize": "13px", + "color": "#666666", + "marginTop": "6px" + } + }, + { + "widgettype": "Text", + "options": { + "label": "一、关联表 pbl_domain_ref 契约（6）", + "fontSize": "16px", + "marginTop": "18px", + "fontWeight": "bold" + } + }, + { + "widgettype": "ResponsableBox", + "options": { + "gap": "16px", + "minWidth": "250px" + }, + "subwidgets": [ { - "type": "button", - "id": "card_ref_list", - "text": "1. 关联列表 pbl_domain_ref_list", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_list.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_list.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "1. 关联记录列表 pbl_domain_ref_list", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "按 tenant_id / ref_type / class_id / team_id 分页查询关联记录", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_ref_get", - "text": "2. 关联详情 pbl_domain_ref_get", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_get.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_get.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "2. 关联记录详情 pbl_domain_ref_get", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "按 id 取单条关联记录（含 ext_json 解析结果）", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_ref_bind", - "text": "3. 绑定 pbl_domain_ref_bind", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_bind.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_bind.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "3. 绑定 pbl_domain_ref_bind", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "幂等建立 ref_type+ref_id 与蓝图/班级/团队的关联", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_ref_update", - "text": "4. 更新 pbl_domain_ref_update", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_update.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_update.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "4. 更新 pbl_domain_ref_update", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "更新 ext_json / blueprint_id / class_id / team_id（租户与归属校验）", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_ref_unbind", - "text": "5. 解绑 pbl_domain_ref_unbind", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_unbind.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_unbind.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "5. 解绑 pbl_domain_ref_unbind", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "删除关联记录（不级联删除基表 world/scene/entity）", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_ref_check_access", - "text": "6. 访问判定 pbl_domain_ref_check_access", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_check_access.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_check_access.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "6. 访问判定 pbl_domain_ref_check_access", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "判定当前租户/班级/团队是否可访问指定 ref_type+ref_id", + "fontSize": "12px", + "color": "#888888" + } + } + ] } ] }, { - "type": "panel", - "id": "grp_world", - "title": "二、world 薄扩展契约（2）", - "layout": "hbox", - "items": [ + "widgettype": "Text", + "options": { + "label": "二、基础域只读查询契约 world / scene / entity（4）", + "fontSize": "16px", + "marginTop": "18px", + "fontWeight": "bold" + } + }, + { + "widgettype": "ResponsableBox", + "options": { + "gap": "16px", + "minWidth": "250px" + }, + "subwidgets": [ { - "type": "button", - "id": "card_world_list", - "text": "7. 租户世界列表 pbl_world_list_by_tenant", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_world_list_by_tenant.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_world_list_by_tenant.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "7. 世界列表 pbl_world_list_by_tenant", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "读基表 world + 关联表过滤，返回本租户可见世界", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_world_ctx", - "text": "8. 世界上下文 pbl_world_get_context", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_world_get_context.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_world_get_context.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "8. 世界上下文 pbl_world_get_context", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "聚合 world 与其下 scene/entity 关联，供运行时取上下文", + "fontSize": "12px", + "color": "#888888" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_scene_list_by_world.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "9. 场景列表 pbl_scene_list_by_world", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "按 world_id 读基表 scene + 关联表过滤", + "fontSize": "12px", + "color": "#888888" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_entity_list_by_scene.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "10. 实体列表 pbl_entity_list_by_scene", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "按 scene_id 读基表 entity + 关联表过滤", + "fontSize": "12px", + "color": "#888888" + } + } + ] } ] }, { - "type": "panel", - "id": "grp_scene_entity", - "title": "三、scene / entity 薄扩展契约（2）", - "layout": "hbox", - "items": [ + "widgettype": "Text", + "options": { + "label": "三、班级 / 团队维度契约（3）", + "fontSize": "16px", + "marginTop": "18px", + "fontWeight": "bold" + } + }, + { + "widgettype": "ResponsableBox", + "options": { + "gap": "16px", + "minWidth": "250px" + }, + "subwidgets": [ { - "type": "button", - "id": "card_scene_list", - "text": "9. 世界下场景 pbl_scene_list_by_world", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_scene_list_by_world.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_team_bind_world.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "11. 团队绑定世界 pbl_team_bind_world", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "写 pbl_domain_ref(ref_type=world, team_id=...)，幂等", + "fontSize": "12px", + "color": "#888888" + } + } + ] }, { - "type": "button", - "id": "card_entity_list", - "text": "10. 场景下实体 pbl_entity_list_by_scene", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_entity_list_by_scene.dspy')}}" + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_team_list_by_class.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "12. 班级下团队 pbl_team_list_by_class", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "按 class_id 聚合关联表，返回团队与其绑定的世界", + "fontSize": "12px", + "color": "#888888" + } + } + ] + }, + { + "widgettype": "VBox", + "options": { + "backgroundColor": "#FFFFFF", + "padding": "20px", + "cursor": "pointer", + "borderRadius": "6px" + }, + "binds": [ + { + "wid": "self", + "event": "click", + "actiontype": "urlwidget", + "target": "app.pbl_domain_ext_content", + "options": { + "url": "{{entire_url('/pbl_domain_ext/api/pbl_team_world_list.dspy')}}?_webbricks_=1" + }, + "mode": "replace" + } + ], + "subwidgets": [ + { + "widgettype": "Text", + "options": { + "label": "13. 团队世界清单 pbl_team_world_list", + "fontWeight": "bold" + } + }, + { + "widgettype": "Text", + "options": { + "label": "按 team_id 返回该团队可访问的世界列表", + "fontSize": "12px", + "color": "#888888" + } + } + ] } ] }, { - "type": "panel", - "id": "grp_team", - "title": "四、班级/团队关联契约（3）", - "layout": "hbox", - "items": [ - { - "type": "button", - "id": "card_team_bind", - "text": "11. 团队绑定世界 pbl_team_bind_world", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_team_bind_world.dspy')}}" - }, - { - "type": "button", - "id": "card_team_world_list", - "text": "12. 团队世界列表 pbl_team_world_list", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_team_world_list.dspy')}}" - }, - { - "type": "button", - "id": "card_team_by_class", - "text": "13. 班级团队列表 pbl_team_list_by_class", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_team_list_by_class.dspy')}}" - } - ] - }, - { - "type": "dataviewer", - "id": "ref_viewer", - "title": "关联记录浏览（CRUD: json/pbl_domain_ref.json）", - "url": "{{entire_url('/pbl_domain_ext/api/pbl_domain_ref_list.dspy')}}" + "widgettype": "VBox", + "id": "pbl_domain_ext_content", + "options": { + "width": "100%", + "flex": "1", + "marginTop": "20px", + "backgroundColor": "#FFFFFF", + "padding": "12px", + "borderRadius": "6px" + } } ] }