# -*- coding: utf-8 -*- """[M8] pbl_domain_ext 契约测试（world/scene/entity 薄扩展）。 运行（环境无 pytest 时用 unittest，二者皆可）:: cd modules/pbl_domain_ext python3 -m unittest discover -s tests -p 'test_*.py' -v python3 -m pytest tests -q # 若环境已装 pytest 覆盖：13 个 dspy 契约 + 1 个物化契约 × 租户隔离(fail-closed) / 基表只读 / 正常 / 异常 四类用例；外加薄扩展铁律（不写基表、只有 1 张新表、列长度口径统一）。 """ import asyncio import os import sys import unittest HERE = os.path.dirname(os.path.abspath(__file__)) ROOT = os.path.dirname(HERE) for _p in (ROOT, HERE): if _p not in sys.path: sys.path.insert(0, _p) import fake_db # noqa: E402 from pbl_domain_ext import api, db, base, errors # noqa: E402 from pbl_domain_ext.base import ( # noqa: E402 FIELD_MAXLEN, LIST_FIELDS, REF_TYPES, TABLE, ) T1 = "T1" # 主测租户（BASE_FIXTURES: world W1/W2, scene S1..S3, entity E1..E3） T2 = "T2" # 他租户（仅 world W3 / scene S9 / entity E9） OP = "U_TESTER" def run(coro): """在无事件循环的 unittest 中执行协程。""" loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro) finally: loop.close() class BaseCase(unittest.TestCase): """每个用例独立内存库，互不污染。""" def setUp(self): self.sor = fake_db.make_fake_sor() db.set_sor(self.sor) def tearDown(self): db.set_sor(None) # 便捷断言 def assert_ok(self, res, msg=""): self.assertTrue(res.get("ok"), f"expected ok, got {res} {msg}") def assert_fail(self, res, code=None, msg=""): self.assertFalse(res.get("ok"), f"expected failure, got {res} {msg}") if code: self.assertEqual(res.get("code"), code, f"wrong code: {res} ({msg})") def refs(self): return self.sor.tables[TABLE] # ============================================================ 关联表契约（写） class TestBind(BaseCase): def test_bind_world_ok(self): res = run(api.pbl_domain_ref_bind( {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "blueprint_id": "BP1", "operator": OP})) self.assert_ok(res) self.assertEqual(res["data"]["action"], "created") ref = res["data"]["ref"] # ref_code / ref_name 自动从基表补全 self.assertEqual(ref["ref_code"], "WLD-001") self.assertEqual(ref["ref_name"], "火星基地") self.assertEqual(ref["bind_state"], "bound") self.assertEqual(len(self.refs()), 1) def test_bind_idempotent_upsert(self): """同 (tenant, ref_type, ref_id, blueprint) 重复绑定 -> 更新而非新增。""" p = {"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "blueprint_id": "BP1", "operator": OP} first = run(api.pbl_domain_ref_bind(p)) self.assert_ok(first) self.assertEqual(first["data"]["action"], "created") second = run(api.pbl_domain_ref_bind(dict(p, ref_name="改名后的世界"))) self.assert_ok(second) self.assertEqual(second["data"]["action"], "updated") self.assertEqual(len(self.refs()), 1, "upsert 必须幂等，不得重复插入") self.assertEqual(self.refs()[0]["ref_name"], "改名后的世界") def test_bind_scene_and_entity(self): for rtype, rid in (("scene", "S1"), ("entity", "E1")): res = run(api.pbl_domain_ref_bind( {"tenant_id": T1, "ref_type": rtype, "ref_id": rid, "blueprint_id": "BP1", "operator": OP})) self.assert_ok(res, rtype) self.assertEqual(len(self.refs()), 2) def test_bind_requires_tenant(self): res = run(api.pbl_domain_ref_bind( {"ref_type": "world", "ref_id": "W1"})) self.assert_fail(res, errors.TENANT_REQUIRED) self.assertEqual(len(self.refs()), 0) def test_bind_invalid_ref_type(self): res = run(api.pbl_domain_ref_bind( {"tenant_id": T1, "ref_type": "player", "ref_id": "W1"})) self.assert_fail(res, errors.REF_TYPE_INVALID) def test_bind_missing_ref_id(self): res = run(api.pbl_domain_ref_bind( {"tenant_id": T1, "ref_type": "world"})) self.assert_fail(res, errors.PARAM_INVALID) def test_bind_cross_tenant_denied(self): """T1 不能绑定 T2 的世界（基表租户可见性校验，fail-closed）。""" res = run(api.pbl_domain_ref_bind( {"tenant_id": T1, "ref_type": "world", "ref_id": "W3"})) self.assert_fail(res, errors.BASE_OBJECT_NOT_FOUND) self.assertEqual(len(self.refs()), 0) self.assertEqual(self.sor.base_write_attempts, [], "校验路径不得写基表") def test_bind_does_not_write_base_tables(self): run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "operator": OP})) self.assertEqual(self.sor.base_write_attempts, []) writes = [c for c in self.sor.calls if c[1] in fake_db.BASE_TABLES and c[0] in ("C", "U", "D", "I")] self.assertEqual(writes, []) class TestUnbind(BaseCase): def test_unbind_soft_delete(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "operator": OP})) self.assert_ok(bound) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_unbind({"tenant_id": T1, "id": rid, "operator": OP})) self.assert_ok(res) self.assertEqual(res["data"]["action"], "unbound") row = self.refs()[0] self.assertEqual(row["is_deleted"], "1", "必须是逻辑删除") self.assertEqual(row["bind_state"], "unbound") self.assertEqual(len(self.refs()), 1, "不得物理删除记录") # 逻辑删除后列表/详情不可见 self.assert_fail(run(api.pbl_domain_ref_get( {"tenant_id": T1, "id": rid})), errors.REF_NOT_FOUND) def test_unbind_idempotent(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "scene", "ref_id": "S1"})) rid = bound["data"]["ref"]["id"] run(api.pbl_domain_ref_unbind({"tenant_id": T1, "id": rid})) again = run(api.pbl_domain_ref_unbind({"tenant_id": T1, "id": rid})) self.assert_ok(again) self.assertEqual(again["data"]["action"], "noop") def test_unbind_requires_tenant_and_id(self): self.assert_fail(run(api.pbl_domain_ref_unbind({"id": "X"})), errors.TENANT_REQUIRED) self.assert_fail(run(api.pbl_domain_ref_unbind({"tenant_id": T1})), errors.PARAM_INVALID) def test_unbind_cross_tenant_denied(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_unbind({"tenant_id": T2, "id": rid})) self.assert_fail(res, errors.REF_NOT_FOUND) self.assertEqual(self.refs()[0]["is_deleted"], "0", "跨租户不得改动他租户记录") class TestUpdate(BaseCase): def test_update_ok(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "operator": OP})) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_update({ "tenant_id": T1, "id": rid, "ref_name": "新名字", "team_id": "TM1", "class_id": "CL1", "remark": "备注", "operator": OP})) self.assert_ok(res) self.assertEqual(res["data"]["ref"]["ref_name"], "新名字") self.assertEqual(res["data"]["ref"]["team_id"], "TM1") self.assertEqual(self.refs()[0]["updater_id"], OP) def test_update_field_length_clipped_per_column(self): """列长度口径统一：blueprint_id=32 / ref_code=64 / ref_name=128， 禁止统一 [:128] 导致 blueprint_id 溢出写库。""" bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_update({ "tenant_id": T1, "id": rid, "blueprint_id": "B" * 100, "ref_code": "C" * 200, "ref_name": "N" * 300, "team_id": "T" * 200, "class_id": "X" * 200})) self.assert_ok(res) ref = res["data"]["ref"] self.assertEqual(len(ref["blueprint_id"]), FIELD_MAXLEN["blueprint_id"]) self.assertEqual(len(ref["ref_code"]), FIELD_MAXLEN["ref_code"]) self.assertEqual(len(ref["ref_name"]), FIELD_MAXLEN["ref_name"]) self.assertEqual(len(ref["team_id"]), FIELD_MAXLEN["team_id"]) self.assertEqual(len(ref["class_id"]), FIELD_MAXLEN["class_id"]) # 真正落库的值同样被截到列宽内 self.assertEqual(len(self.refs()[0]["blueprint_id"]), 32) def test_bind_field_length_clipped_per_column(self): """bind 与 update 共用 FIELD_MAXLEN 口径（两处一致）。""" res = run(api.pbl_domain_ref_bind({ "tenant_id": T1, "ref_type": "world", "ref_id": "W2", "blueprint_id": "B" * 100, "team_id": "T" * 200, "class_id": "X" * 200, "ref_name": "N" * 300})) self.assert_ok(res) ref = res["data"]["ref"] self.assertEqual(len(ref["blueprint_id"]), 32) self.assertEqual(len(ref["team_id"]), 64) self.assertEqual(len(ref["class_id"]), 64) self.assertEqual(len(ref["ref_name"]), 128) def test_update_rejects_unknown_bind_state(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) res = run(api.pbl_domain_ref_update({"tenant_id": T1, "id": bound["data"]["ref"]["id"], "bind_state": "frozen"})) self.assert_fail(res, errors.PARAM_INVALID) def test_update_noop_when_no_changes(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_update({"tenant_id": T1, "id": rid})) self.assert_ok(res) self.assertEqual(res["data"]["action"], "noop") def test_update_cross_tenant_denied(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_update({"tenant_id": T2, "id": rid, "ref_name": "越权改名"})) self.assert_fail(res, errors.REF_NOT_FOUND) self.assertEqual(self.refs()[0]["ref_name"], "火星基地") def test_update_requires_tenant(self): self.assert_fail(run(api.pbl_domain_ref_update({"id": "X"})), errors.TENANT_REQUIRED) class TestGetAndList(BaseCase): def test_get_ok_and_shape(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) rid = bound["data"]["ref"]["id"] res = run(api.pbl_domain_ref_get({"tenant_id": T1, "id": rid})) self.assert_ok(res) self.assertEqual(sorted(res["data"].keys()), sorted(LIST_FIELDS)) def test_get_cross_tenant_denied(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) self.assert_fail(run(api.pbl_domain_ref_get( {"tenant_id": T2, "id": bound["data"]["ref"]["id"]})), errors.REF_NOT_FOUND) def test_list_pagination_and_filter(self): for rid in ("W1", "W2"): run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": rid, "blueprint_id": "BP1"})) run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "scene", "ref_id": "S1", "blueprint_id": "BP1"})) res = run(api.pbl_domain_ref_list({"tenant_id": T1, "page": 1, "rows": 2})) self.assert_ok(res) self.assertEqual(res["data"]["total"], 3) self.assertEqual(len(res["data"]["rows"]), 2) filtered = run(api.pbl_domain_ref_list({"tenant_id": T1, "ref_type": "world"})) self.assertEqual(filtered["data"]["total"], 2) by_bp = run(api.pbl_domain_ref_list({"tenant_id": T1, "blueprint_id": "BP1", "bind_state": "bound"})) self.assertEqual(by_bp["data"]["total"], 3) def test_list_invalid_ref_type(self): res = run(api.pbl_domain_ref_list({"tenant_id": T1, "ref_type": "bogus"})) self.assert_fail(res, errors.REF_TYPE_INVALID) def test_list_requires_tenant(self): self.assert_fail(run(api.pbl_domain_ref_list({})), errors.TENANT_REQUIRED) def test_list_isolated_per_tenant(self): run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) res = run(api.pbl_domain_ref_list({"tenant_id": T2})) self.assert_ok(res) self.assertEqual(res["data"]["total"], 0) class TestCheckAccess(BaseCase): def test_check_access_by_id(self): bound = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1"})) rid = bound["data"]["ref"]["id"] self.assertTrue(run(api.pbl_domain_ref_check_access( {"tenant_id": T1, "id": rid}))["data"]["allowed"]) self.assertFalse(run(api.pbl_domain_ref_check_access( {"tenant_id": T2, "id": rid}))["data"]["allowed"]) def test_check_access_by_object(self): run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "scene", "ref_id": "S2"})) self.assertTrue(run(api.pbl_domain_ref_check_access( {"tenant_id": T1, "ref_type": "scene", "ref_id": "S2"})) ["data"]["allowed"]) self.assertFalse(run(api.pbl_domain_ref_check_access( {"tenant_id": T1, "ref_type": "scene", "ref_id": "S3"})) ["data"]["allowed"]) def test_check_access_fail_closed_without_params(self): res = run(api.pbl_domain_ref_check_access({"tenant_id": T1})) self.assert_ok(res) self.assertFalse(res["data"]["allowed"]) self.assert_fail(run(api.pbl_domain_ref_check_access({})), errors.TENANT_REQUIRED) # ============================================================ 基表只读视图契约 class TestBaseViews(BaseCase): def test_world_list_by_tenant(self): res = run(api.pbl_world_list_by_tenant({"tenant_id": T1})) self.assert_ok(res) ids = sorted(r["id"] for r in res["data"]["rows"]) self.assertEqual(ids, ["W1", "W2"]) self.assertEqual(res["data"]["total"], 2) def test_world_list_cross_tenant_isolated(self): res = run(api.pbl_world_list_by_tenant({"tenant_id": T2})) self.assertEqual([r["id"] for r in res["data"]["rows"]], ["W3"]) def test_world_get_context_ok(self): res = run(api.pbl_world_get_context({"tenant_id": T1, "world_id": "W1"})) self.assert_ok(res) data = res["data"] self.assertEqual(data["world"]["name"], "火星基地") self.assertEqual(data["scene_count"], 2) self.assertEqual(data["entity_count"], 3) s1 = [s for s in data["scenes"] if s["id"] == "S1"][0] self.assertEqual(s1["entity_count"], 2) self.assertEqual(sorted(e["id"] for e in s1["entities"]), ["E1", "E2"]) def test_world_get_context_missing(self): self.assert_fail(run(api.pbl_world_get_context( {"tenant_id": T1, "world_id": "W999"})), errors.BASE_OBJECT_NOT_FOUND) self.assert_fail(run(api.pbl_world_get_context( {"tenant_id": T1})), errors.PARAM_INVALID) def test_world_get_context_cross_tenant(self): self.assert_fail(run(api.pbl_world_get_context( {"tenant_id": T1, "world_id": "W3"})), errors.BASE_OBJECT_NOT_FOUND) def test_scene_list_by_world(self): res = run(api.pbl_scene_list_by_world({"tenant_id": T1, "world_id": "W1"})) self.assert_ok(res) self.assertEqual(sorted(r["id"] for r in res["data"]["rows"]), ["S1", "S2"]) self.assert_fail(run(api.pbl_scene_list_by_world({"tenant_id": T1})), errors.PARAM_INVALID) def test_entity_list_by_scene(self): res = run(api.pbl_entity_list_by_scene({"tenant_id": T1, "scene_id": "S1"})) self.assert_ok(res) self.assertEqual(sorted(r["id"] for r in res["data"]["rows"]), ["E1", "E2"]) other = run(api.pbl_entity_list_by_scene({"tenant_id": T2, "scene_id": "S1"})) self.assertEqual(other["data"]["total"], 0) def test_base_views_require_tenant(self): for coro in (api.pbl_world_list_by_tenant({}), api.pbl_world_get_context({"world_id": "W1"}), api.pbl_scene_list_by_world({"world_id": "W1"}), api.pbl_entity_list_by_scene({"scene_id": "S1"})): self.assert_fail(run(coro), errors.TENANT_REQUIRED) def test_base_views_never_write_base_tables(self): run(api.pbl_world_list_by_tenant({"tenant_id": T1})) run(api.pbl_world_get_context({"tenant_id": T1, "world_id": "W1"})) run(api.pbl_scene_list_by_world({"tenant_id": T1, "world_id": "W1"})) run(api.pbl_entity_list_by_scene({"tenant_id": T1, "scene_id": "S1"})) self.assertEqual(self.sor.base_write_attempts, []) # ============================================================ 团队 × 世界契约 class TestTeamWorld(BaseCase): def test_team_bind_world_and_list(self): res = run(api.pbl_team_bind_world({"tenant_id": T1, "team_id": "TM1", "class_id": "CL1", "world_id": "W1", "operator": OP})) self.assert_ok(res) self.assertEqual(res["data"]["ref"]["ref_type"], "world") listed = run(api.pbl_team_world_list({"tenant_id": T1, "team_id": "TM1"})) self.assert_ok(listed) self.assertEqual(listed["data"]["total"], 1) row = listed["data"]["rows"][0] self.assertEqual(row["world"]["id"], "W1") self.assertEqual(row["class_id"], "CL1") def test_team_world_list_skips_invisible_base_object(self): """关联存在但基表对象不可见（世界被删）→ 不呈现（fail-closed）。""" run(api.pbl_team_bind_world({"tenant_id": T1, "team_id": "TM1", "world_id": "W1"})) self.sor.base["world"] = [w for w in self.sor.base["world"] if w["id"] != "W1"] listed = run(api.pbl_team_world_list({"tenant_id": T1, "team_id": "TM1"})) self.assertEqual(listed["data"]["total"], 0) def test_team_world_list_requires_team(self): self.assert_fail(run(api.pbl_team_world_list({"tenant_id": T1})), errors.PARAM_INVALID) self.assert_fail(run(api.pbl_team_world_list({"team_id": "TM1"})), errors.TENANT_REQUIRED) def test_team_world_list_isolated_per_tenant(self): run(api.pbl_team_bind_world({"tenant_id": T1, "team_id": "TM1", "world_id": "W1"})) res = run(api.pbl_team_world_list({"tenant_id": T2, "team_id": "TM1"})) self.assertEqual(res["data"]["total"], 0) def test_team_list_by_class_group_counts(self): for team, world in (("TM1", "W1"), ("TM1", "W2"), ("TM2", "W1")): run(api.pbl_team_bind_world({"tenant_id": T1, "team_id": team, "class_id": "CL1", "world_id": world})) res = run(api.pbl_team_list_by_class({"tenant_id": T1, "class_id": "CL1"})) self.assert_ok(res) counts = {r["team_id"]: r["world_count"] for r in res["data"]["rows"]} self.assertEqual(counts, {"TM1": 2, "TM2": 1}) self.assertEqual(res["data"]["total"], 2) self.assert_fail(run(api.pbl_team_list_by_class({"tenant_id": T1})), errors.PARAM_INVALID) def test_unbound_team_world_not_counted(self): run(api.pbl_team_bind_world({"tenant_id": T1, "team_id": "TM1", "class_id": "CL1", "world_id": "W1"})) bound = run(api.pbl_team_world_list({"tenant_id": T1, "team_id": "TM1"})) rid = bound["data"]["rows"][0]["id"] run(api.pbl_domain_ref_unbind({"tenant_id": T1, "id": rid})) res = run(api.pbl_team_list_by_class({"tenant_id": T1, "class_id": "CL1"})) self.assertEqual(res["data"]["rows"], []) # ============================================================ 物化契约（pbl_compiler） class TestMaterialize(BaseCase): def test_materialize_groups_by_ref_type(self): run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "blueprint_id": "BP1"})) run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "scene", "ref_id": "S1", "blueprint_id": "BP1"})) run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "entity", "ref_id": "E1", "blueprint_id": "BP1"})) run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W2", "blueprint_id": "BP2"})) res = run(api.pbl_domain_materialize_game_definition( {"tenant_id": T1, "blueprint_id": "BP1"})) self.assert_ok(res) data = res["data"] self.assertEqual(data["counts"], {"world": 1, "scene": 1, "entity": 1}) self.assertEqual(len(data["worlds"]), 1) self.assertEqual(data["worlds"][0]["base"]["id"], "W1") self.assertEqual(data["table"], TABLE) def test_materialize_excludes_unbound_and_invisible(self): b1 = run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W1", "blueprint_id": "BP1"})) run(api.pbl_domain_ref_bind({"tenant_id": T1, "ref_type": "world", "ref_id": "W2", "blueprint_id": "BP1"})) run(api.pbl_domain_ref_unbind({"tenant_id": T1, "id": b1["data"]["ref"]["id"]})) res = run(api.pbl_domain_materialize_game_definition( {"tenant_id": T1, "blueprint_id": "BP1"})) self.assert_ok(res) self.assertEqual(res["data"]["counts"]["world"], 1) self.assertEqual(res["data"]["worlds"][0]["ref_id"], "W2") def test_materialize_requires_tenant(self): self.assert_fail(run(api.pbl_domain_materialize_game_definition({})), errors.TENANT_REQUIRED) def test_materialize_symbol_is_exported(self): """import 闭包：pbl_compiler 以 from pbl_domain_ext.api import ... 引用。""" self.assertTrue(callable( getattr(api, "pbl_domain_materialize_game_definition"))) import pbl_domain_ext as pkg self.assertTrue(hasattr(pkg, "pbl_domain_materialize_game_definition")) self.assertTrue(hasattr(pkg, "load_pbl_domain_ext")) # ============================================================ 薄扩展铁律 class TestThinExtensionInvariants(BaseCase): def test_only_one_new_table(self): """本模块只新增 1 张关联表，不为 world/scene/entity 各建一张。""" self.assertEqual(TABLE, "pbl_domain_ref") self.assertEqual(len(base.BASE_TABLES), 3) for wrong in ("pbl_world_ref", "pbl_scene_ref", "pbl_entity_ref"): self.assertNotIn(wrong, self.sor.tables) def test_ref_types_match_design(self): self.assertEqual(REF_TYPES, ("world", "scene", "entity")) self.assertEqual(sorted(REF_TYPES), sorted(base.BASE_TABLES)) def test_ext_json_is_authoritative_column(self): """设计 §J1 权威列名 ext_json 必须在模型列中存在，旧名不得残留。""" import json with open(os.path.join(ROOT, "models", "pbl_domain_ref.json"), encoding="utf-8") as fp: model = json.load(fp) cols = [f["name"] for f in model["fields"]] self.assertIn("ext_json", cols) self.assertNotIn("domain_ext_json", cols) def test_clip_respects_field_maxlen(self): self.assertEqual(base.clip("blueprint_id", "x" * 99), "x" * 32) self.assertEqual(base.clip("ref_name", "y" * 999), "y" * 128) self.assertEqual(base.clip("remark", "z" * 999), "z" * 255) def test_db_layer_rejects_base_table_writes(self): """db 层 + FakeSor 双保险：任何指向基表的写操作必须抛错并被记录。""" with self.assertRaises(AssertionError): db.assert_writable("world") for table in ("world", "scene", "entity"): with self.assertRaises(AssertionError): run(self.sor.C(table, {"id": "X", "tenant_id": T1})) with self.assertRaises(AssertionError): run(self.sor.U(table, {"data": {"name": "x"}, "where": {"id": "X"}})) with self.assertRaises(AssertionError): run(self.sor.D(table, {"where": {"id": "X"}})) self.assertEqual([op for op, _t in self.sor.base_write_attempts if op in ("C", "U", "D", "I")], ["C", "U", "D"] * 3, "写基表尝试必须全部被记录，供测试断言") # db 层的写函数只允许指向 pbl_domain_ref：直接以基表名调用底层适配器必须抛错 for table in ("world", "scene", "entity"): with self.assertRaises(AssertionError): db._sor_c(self.sor, table, {"id": "X"}) with self.assertRaises(AssertionError): db._sor_u(self.sor, table, {"name": "x"}, {"id": "X"}) with self.assertRaises(AssertionError): db.assert_writable(table) self.assertTrue(callable(db.insert_ref)) self.assertEqual(db.assert_writable(db.TABLE), None, "写本模块表必须放行") def test_read_base_whitelist(self): rows = run(db.read_base("world", {"tenant_id": T1})) self.assertEqual(sorted(r["id"] for r in rows), ["W1", "W2"]) with self.assertRaises(ValueError): run(db.read_base("no_such_table", {})) with self.assertRaises(ValueError): run(db.read_base("pbl_domain_ref", {})) def test_no_hardcoded_dbname(self): """db.py 不得出现 DBNAME 常量（库名由宿主 get_module_dbname 决定）。""" with open(os.path.join(ROOT, "pbl_domain_ext", "db.py"), encoding="utf-8") as fp: src = fp.read() self.assertNotIn("DBNAME =", src) self.assertIn("get_module_dbname", src) for fake in ("sor.save", "sor.list", "sor.insert", "sor.query"): self.assertNotIn(fake + "(", src) if __name__ == "__main__": unittest.main(verbosity=2)