596 lines
27 KiB
Python
596 lines
27 KiB
Python
# -*- 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]["updated_by"], 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
|
||
with self.assertRaises(AssertionError):
|
||
run(db.insert_ref({"id": "X"})) if False else None
|
||
self.assertTrue(callable(db.insert_ref))
|
||
|
||
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)
|