449 lines
22 KiB
Python
449 lines
22 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 契约对应的 api 函数 × 租户隔离 / 权限(只读基表) / 正常 / 异常 四类用例，
|
||
外加薄扩展铁律（不写基表、表总账只有 pbl_domain_ref 一张新表、ext_json 字段名对齐设计 §J1）。
|
||
"""
|
||
|
||
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 # noqa: E402
|
||
from pbl_domain_ext.base import EXT_FIELD, REF_TYPES, TABLE # noqa: E402
|
||
|
||
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.sor = fake_db.make_fake_sor()
|
||
db.set_sor(self.sor)
|
||
|
||
def tearDown(self):
|
||
db.set_sor(None)
|
||
|
||
# ---- 断言助手 -------------------------------------------------------
|
||
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, []))
|
||
|
||
|
||
# ==========================================================================
|
||
# 1. 关联表 pbl_domain_ref：绑定 / 解绑 / 更新 / 查询
|
||
# ==========================================================================
|
||
class TestBind(BaseCase):
|
||
|
||
def test_bind_world_ok(self):
|
||
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_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_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_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_rejects_bad_ref_type(self):
|
||
self.assertErr(self.bind("course", "W1"), "PBL_DE_REF_TYPE_INVALID",
|
||
"ref_type 仅允许 world/scene/entity")
|
||
|
||
def test_bind_rejects_empty_ref_id(self):
|
||
self.assertErr(self.bind("world", ""), "PBL_DE_PARAM_INVALID", "ref_id 为空")
|
||
|
||
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_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_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 TestUnbindUpdateGet(BaseCase):
|
||
|
||
def test_unbind_soft_delete(self):
|
||
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.assertErr(api.pbl_domain_ref_unbind(
|
||
{"tenant_id": T1, "ref_type": "world", "ref_id": "W1"}),
|
||
"PBL_DE_NOT_FOUND", "解绑不存在的关联")
|
||
|
||
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_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_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_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_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 TestCheckAccess(BaseCase):
|
||
|
||
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_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_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_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")
|
||
|
||
|
||
# ==========================================================================
|
||
# 2. 基础域只读契约（world / scene / entity）
|
||
# ==========================================================================
|
||
class TestBaseDomainReadOnly(BaseCase):
|
||
|
||
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_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_world_list_requires_tenant(self):
|
||
self.assertErr(api.pbl_world_list_by_tenant({}), "PBL_DE_TENANT_MISSING")
|
||
|
||
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_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_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_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_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_scene_list_requires_world_id(self):
|
||
self.assertErr(api.pbl_scene_list_by_world({"tenant_id": T1}),
|
||
"PBL_DE_PARAM_INVALID")
|
||
|
||
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_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_entity_list_requires_scene_id(self):
|
||
self.assertErr(api.pbl_entity_list_by_scene({"tenant_id": T1}),
|
||
"PBL_DE_PARAM_INVALID")
|
||
|
||
|
||
# ==========================================================================
|
||
# 3. 班级 / 团队维度契约
|
||
# ==========================================================================
|
||
class TestTeamClass(BaseCase):
|
||
|
||
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_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_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_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_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_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_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_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_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_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_ref_types_cover_three_domains(self):
|
||
self.assertEqual(tuple(REF_TYPES), ("world", "scene", "entity"))
|
||
|
||
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_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_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_materialize_game_definition_exported(self):
|
||
"""历史 QC 硬门禁：pbl_compiler 曾引用该符号，必须真实存在且可调用。"""
|
||
import pbl_domain_ext as pkg
|
||
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__":
|
||
unittest.main(verbosity=2)
|