515 lines
23 KiB
Python
515 lines
23 KiB
Python
"""M8 基础域薄扩展离线单测（不依赖 Sage 运行时 / 不连库）。
|
||
|
||
运行：cd modules/pbl_domain_ext && python3 -m pytest tests/ -q
|
||
或 python3 tests/test_assoc.py（无 pytest 时直接跑）
|
||
|
||
覆盖：租户强制打头 / 幂等 upsert / 可见性推导与校验 / 过滤白名单 /
|
||
作用域回溯解析 / Game Definition 物化 / 不改基表铁律 / 契约层错误码。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import unittest
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
from fake_db import FakeDb # noqa: E402 (tests/ 目录内导入)
|
||
|
||
from pbl_domain_ext import api, assoc # noqa: E402
|
||
from pbl_domain_ext.assoc import DomainExtError # noqa: E402
|
||
|
||
TENANT = "T-DEMO-001"
|
||
|
||
|
||
def new_db() -> FakeDb:
|
||
return FakeDb([assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF])
|
||
|
||
|
||
class TestTenantGuard(unittest.TestCase):
|
||
"""租户强制打头：缺失/空/超长一律 fail-closed。"""
|
||
|
||
def test_missing_tenant_raises(self):
|
||
db = new_db()
|
||
for bad in (None, "", " "):
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.upsert_world_ref(db, bad, 101)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED)
|
||
|
||
def test_overlong_tenant_raises(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.upsert_world_ref(db, "T" * 65, 101)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED)
|
||
|
||
def test_missing_key_raises(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.upsert_world_ref(db, TENANT, None)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED)
|
||
|
||
def test_non_integer_key_raises(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_scene_ref(db, TENANT, "abc")
|
||
|
||
def test_negative_key_raises(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_entity_ref(db, TENANT, -3)
|
||
|
||
def test_list_without_tenant_raises(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.list_world_refs(db, None)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_TENANT_REQUIRED)
|
||
|
||
|
||
class TestUpsertIdempotent(unittest.TestCase):
|
||
def test_create_then_update_same_row(self):
|
||
db = new_db()
|
||
first = assoc.upsert_world_ref(db, TENANT, 101, class_id=7, team_id=9, role_code="teacher")
|
||
self.assertEqual(first["action"], "created")
|
||
self.assertGreater(first["id"], 0)
|
||
second = assoc.upsert_world_ref(db, TENANT, 101, class_id=8, team_id=9, role_code="teacher")
|
||
self.assertEqual(second["action"], "updated")
|
||
self.assertEqual(second["id"], first["id"])
|
||
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
|
||
self.assertEqual(db.tables[assoc.WORLD_REF][0]["class_id"], 8)
|
||
|
||
def test_different_role_is_different_row(self):
|
||
db = new_db()
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code="teacher")
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code="student")
|
||
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 2)
|
||
|
||
def test_different_tenant_isolated(self):
|
||
db = new_db()
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9)
|
||
assoc.upsert_world_ref(db, "T-OTHER", 101, team_id=9)
|
||
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 2)
|
||
page = assoc.list_world_refs(db, TENANT)
|
||
self.assertEqual(page["total"], 1)
|
||
self.assertEqual(page["items"][0]["tenant_id"], TENANT)
|
||
|
||
def test_scene_and_entity_parents_persisted(self):
|
||
db = new_db()
|
||
assoc.upsert_scene_ref(db, TENANT, 202, world_id=101, team_id=9)
|
||
assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
|
||
scene_row = db.tables[assoc.SCENE_REF][0]
|
||
entity_row = db.tables[assoc.ENTITY_REF][0]
|
||
self.assertEqual(scene_row["world_id"], 101)
|
||
self.assertEqual(entity_row["scene_id"], 202)
|
||
self.assertEqual(entity_row["world_id"], 101)
|
||
|
||
def test_unknown_parent_column_rejected(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_ref(db, assoc.SCENE_REF, TENANT, 202, {"entity_id": 1})
|
||
|
||
def test_ext_json_serialized(self):
|
||
db = new_db()
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, ext={"stage": 2, "tags": ["a", "b"]})
|
||
stored = db.tables[assoc.WORLD_REF][0]["ext_json"]
|
||
self.assertEqual(json.loads(stored)["stage"], 2)
|
||
|
||
def test_ext_string_json_passthrough(self):
|
||
db = new_db()
|
||
assoc.upsert_world_ref(db, TENANT, 102, team_id=9, ext='{"k": 1}')
|
||
self.assertEqual(json.loads(db.tables[assoc.WORLD_REF][0]["ext_json"])["k"], 1)
|
||
|
||
def test_ext_bad_json_rejected(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, ext="{not json")
|
||
|
||
def test_bad_source_rejected(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, source="drop table;")
|
||
|
||
|
||
class TestVisibility(unittest.TestCase):
|
||
def test_auto_visibility_by_scope(self):
|
||
db = new_db()
|
||
self.assertEqual(assoc.upsert_world_ref(db, TENANT, 1, team_id=5)["visibility"], "team")
|
||
self.assertEqual(assoc.upsert_world_ref(db, TENANT, 2, class_id=5)["visibility"], "class")
|
||
self.assertEqual(assoc.upsert_world_ref(db, TENANT, 3)["visibility"], "tenant")
|
||
|
||
def test_explicit_visibility_must_match_scope(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.upsert_world_ref(db, TENANT, 4, visibility="team")
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED)
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_world_ref(db, TENANT, 5, visibility="class")
|
||
|
||
def test_bad_visibility_value(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.upsert_world_ref(db, TENANT, 6, visibility="public")
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_BAD_VISIBILITY)
|
||
|
||
|
||
class TestListAndFilter(unittest.TestCase):
|
||
def setUp(self):
|
||
self.db = new_db()
|
||
for i in range(1, 8):
|
||
assoc.upsert_world_ref(self.db, TENANT, 100 + i, class_id=7, team_id=9)
|
||
assoc.upsert_world_ref(self.db, "T-OTHER", 999, team_id=1)
|
||
|
||
def test_pagination(self):
|
||
page = assoc.list_world_refs(self.db, TENANT, limit=3, offset=0)
|
||
self.assertEqual(page["total"], 7)
|
||
self.assertEqual(len(page["items"]), 3)
|
||
page2 = assoc.list_world_refs(self.db, TENANT, limit=3, offset=6)
|
||
self.assertEqual(len(page2["items"]), 1)
|
||
|
||
def test_limit_clamped(self):
|
||
page = assoc.list_world_refs(self.db, TENANT, limit=99999)
|
||
self.assertEqual(page["limit"], assoc.MAX_LIMIT)
|
||
|
||
def test_filter_whitelist(self):
|
||
page = assoc.list_world_refs(self.db, TENANT, {"class_id": 7})
|
||
self.assertEqual(page["total"], 7)
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.list_world_refs(self.db, TENANT, {"password": "x"})
|
||
|
||
def test_empty_filter_value_ignored(self):
|
||
page = assoc.list_world_refs(self.db, TENANT, {"role_code": "", "team_id": 9})
|
||
self.assertEqual(page["total"], 7)
|
||
|
||
def test_scene_filter_by_world(self):
|
||
assoc.upsert_scene_ref(self.db, TENANT, 201, world_id=101, team_id=9)
|
||
assoc.upsert_scene_ref(self.db, TENANT, 202, world_id=102, team_id=9)
|
||
page = assoc.list_scene_refs(self.db, TENANT, {"world_id": 101})
|
||
self.assertEqual(page["total"], 1)
|
||
self.assertEqual(page["items"][0]["scene_id"], 201)
|
||
|
||
|
||
class TestGetDelete(unittest.TestCase):
|
||
def test_get_strict_and_lenient(self):
|
||
db = new_db()
|
||
assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
|
||
row = assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 303, team_id=9)
|
||
self.assertEqual(row["scene_id"], 202)
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 404, team_id=9)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_NOT_FOUND)
|
||
self.assertIsNone(assoc.get_ref(db, assoc.ENTITY_REF, TENANT, 404, team_id=9, strict=False))
|
||
|
||
def test_delete_only_own_tenant(self):
|
||
db = new_db()
|
||
res = assoc.upsert_world_ref(db, TENANT, 101, team_id=9)
|
||
assoc.upsert_world_ref(db, "T-OTHER", 101, team_id=9)
|
||
affected = assoc.delete_ref(db, assoc.WORLD_REF, TENANT, res["id"])
|
||
self.assertEqual(affected, 1)
|
||
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
|
||
self.assertEqual(db.tables[assoc.WORLD_REF][0]["tenant_id"], "T-OTHER")
|
||
|
||
|
||
class TestScopeResolve(unittest.TestCase):
|
||
def setUp(self):
|
||
self.db = new_db()
|
||
assoc.upsert_world_ref(self.db, TENANT, 101, class_id=7, team_id=9)
|
||
assoc.upsert_scene_ref(self.db, TENANT, 202, world_id=101, class_id=7, team_id=9)
|
||
assoc.upsert_entity_ref(self.db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
|
||
|
||
def test_resolve_from_entity_backfills_parents(self):
|
||
scope = assoc.resolve_scope(self.db, TENANT, entity_id=303)
|
||
self.assertEqual(scope["resolved_from"], "entity")
|
||
self.assertEqual(scope["scene_id"], 202)
|
||
self.assertEqual(scope["world_id"], 101)
|
||
self.assertEqual(scope["team_id"], 9)
|
||
self.assertEqual(scope["class_id"], 7)
|
||
self.assertEqual(scope["visibility"], "team")
|
||
self.assertIsNotNone(scope["world"])
|
||
|
||
def test_resolve_from_scene_only(self):
|
||
scope = assoc.resolve_scope(self.db, TENANT, scene_id=202)
|
||
self.assertEqual(scope["resolved_from"], "scene")
|
||
self.assertEqual(scope["world_id"], 101)
|
||
|
||
def test_resolve_from_world_only(self):
|
||
scope = assoc.resolve_scope(self.db, TENANT, world_id=101)
|
||
self.assertEqual(scope["resolved_from"], "world")
|
||
self.assertEqual(scope["class_id"], 7)
|
||
|
||
def test_resolve_requires_one_id(self):
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.resolve_scope(self.db, TENANT)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_PARAM_REQUIRED)
|
||
|
||
def test_resolve_not_found_fail_closed(self):
|
||
with self.assertRaises(DomainExtError) as ctx:
|
||
assoc.resolve_scope(self.db, TENANT, entity_id=999)
|
||
self.assertEqual(ctx.exception.code, assoc.ERR_NOT_FOUND)
|
||
|
||
def test_resolve_cross_tenant_not_leaked(self):
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.resolve_scope(self.db, "T-OTHER", entity_id=303)
|
||
|
||
def test_resolve_entity_without_scene_parent(self):
|
||
assoc.upsert_entity_ref(self.db, TENANT, 404, world_id=101, team_id=9)
|
||
scope = assoc.resolve_scope(self.db, TENANT, entity_id=404)
|
||
self.assertEqual(scope["resolved_from"], "entity")
|
||
self.assertEqual(scope["world_id"], 101)
|
||
self.assertIsNone(scope["scene"])
|
||
|
||
|
||
class TestMaterialize(unittest.TestCase):
|
||
DEFINITION = {
|
||
"game": {
|
||
"world_id": 101,
|
||
"scenes": [{"scene_id": 202, "entities": [{"entity_id": 303}, {"entityId": 304}]}],
|
||
},
|
||
"meta": {"worldId": 101},
|
||
}
|
||
|
||
def test_collect_ids_dedup(self):
|
||
ids = assoc.collect_ids(self.DEFINITION, assoc.WORLD_REF)
|
||
self.assertEqual(ids, [101])
|
||
self.assertEqual(assoc.collect_ids(self.DEFINITION, assoc.SCENE_REF), [202])
|
||
self.assertEqual(sorted(assoc.collect_ids(self.DEFINITION, assoc.ENTITY_REF)), [303, 304])
|
||
|
||
def test_collect_ids_from_json_string(self):
|
||
ids = assoc.collect_ids(json.dumps(self.DEFINITION), assoc.WORLD_REF)
|
||
self.assertEqual(ids, [101])
|
||
|
||
def test_bind_creates_all_refs(self):
|
||
db = new_db()
|
||
summary = assoc.bind_game_definition(db, TENANT, self.DEFINITION, class_id=7, team_id=9)
|
||
self.assertEqual(summary["world_refs"], 1)
|
||
self.assertEqual(summary["scene_refs"], 1)
|
||
self.assertEqual(summary["entity_refs"], 2)
|
||
self.assertEqual(summary["created"], 4)
|
||
self.assertEqual(summary["source"], "compiler")
|
||
self.assertEqual(len(db.tables[assoc.ENTITY_REF]), 2)
|
||
|
||
def test_bind_is_idempotent(self):
|
||
db = new_db()
|
||
assoc.bind_game_definition(db, TENANT, self.DEFINITION, team_id=9)
|
||
second = assoc.bind_game_definition(db, TENANT, json.dumps(self.DEFINITION), team_id=9)
|
||
self.assertEqual(second["updated"], 4)
|
||
self.assertEqual(second["created"], 0)
|
||
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
|
||
|
||
def test_bind_rejects_empty_and_bad_json(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.bind_game_definition(db, TENANT, " ")
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.bind_game_definition(db, TENANT, "{bad json")
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.bind_game_definition(db, TENANT, [1, 2, 3])
|
||
|
||
def test_bind_without_ids_reports_skip(self):
|
||
db = new_db()
|
||
summary = assoc.bind_game_definition(db, TENANT, {"game": {"name": "x"}})
|
||
self.assertEqual(summary["world_refs"], 0)
|
||
self.assertTrue(summary["skipped"])
|
||
|
||
def test_multiple_scenes_leave_parent_zero(self):
|
||
db = new_db()
|
||
definition = {"scenes": [{"scene_id": 1}, {"scene_id": 2}], "entities": [{"entity_id": 5}]}
|
||
summary = assoc.bind_game_definition(db, TENANT, definition, team_id=9)
|
||
self.assertEqual(summary["scene_refs"], 2)
|
||
self.assertEqual(db.tables[assoc.ENTITY_REF][0]["scene_id"], 0)
|
||
self.assertTrue(any("多个 scene_id" in s for s in summary["skipped"]))
|
||
|
||
def test_multiple_worlds_take_first(self):
|
||
db = new_db()
|
||
definition = {"worlds": [{"world_id": 11}, {"world_id": 22}], "scenes": [{"scene_id": 5}]}
|
||
summary = assoc.bind_game_definition(db, TENANT, definition, team_id=9)
|
||
self.assertEqual(summary["world_refs"], 2)
|
||
self.assertEqual(db.tables[assoc.SCENE_REF][0]["world_id"], 11)
|
||
self.assertTrue(any("多个 world_id" in s for s in summary["skipped"]))
|
||
|
||
|
||
class TestBaseTableUntouched(unittest.TestCase):
|
||
"""铁律：任何路径都不得写 world/scene/entity 基表。"""
|
||
|
||
def test_no_base_table_write(self):
|
||
db = new_db()
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9)
|
||
assoc.upsert_scene_ref(db, TENANT, 202, world_id=101, team_id=9)
|
||
assoc.upsert_entity_ref(db, TENANT, 303, scene_id=202, world_id=101, team_id=9)
|
||
assoc.bind_game_definition(db, TENANT, TestMaterialize.DEFINITION, team_id=9)
|
||
assoc.resolve_scope(db, TENANT, entity_id=303)
|
||
self.assertEqual(db.base_table_writes, [])
|
||
for sql, _params in db.executed:
|
||
lowered = " ".join(sql.lower().split())
|
||
if lowered.startswith(("insert", "update", "delete")):
|
||
for base in ("world", "scene", "entity"):
|
||
self.assertFalse(lowered.startswith("insert into %s " % base), sql)
|
||
self.assertFalse(lowered.startswith("update %s " % base), sql)
|
||
self.assertFalse(lowered.startswith("delete from %s " % base), sql)
|
||
|
||
def test_all_writes_target_ref_tables_only(self):
|
||
db = new_db()
|
||
assoc.bind_game_definition(db, TENANT, TestMaterialize.DEFINITION, team_id=9)
|
||
write_tables = set()
|
||
for sql, _params in db.executed:
|
||
lowered = " ".join(sql.lower().split())
|
||
if lowered.startswith("insert into"):
|
||
write_tables.add(lowered.split()[2])
|
||
elif lowered.startswith("update"):
|
||
write_tables.add(lowered.split()[1])
|
||
elif lowered.startswith("delete from"):
|
||
write_tables.add(lowered.split()[2])
|
||
self.assertTrue(write_tables)
|
||
self.assertTrue(write_tables.issubset(set(assoc.REF_TABLES)), write_tables)
|
||
|
||
def test_table_name_injection_guard(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.list_refs(db, "pbl_world_ref; DROP TABLE world", TENANT)
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.upsert_ref(db, "world", TENANT, 1)
|
||
|
||
def test_filter_column_injection_guard(self):
|
||
db = new_db()
|
||
with self.assertRaises(DomainExtError):
|
||
assoc.list_refs(db, assoc.WORLD_REF, TENANT, {"1=1 OR tenant_id": "x"})
|
||
|
||
def test_values_always_parameterized(self):
|
||
db = new_db()
|
||
evil = "x'; DROP TABLE pbl_world_ref; --"
|
||
assoc.upsert_world_ref(db, TENANT, 101, team_id=9, role_code=evil)
|
||
for sql, params in db.executed:
|
||
self.assertNotIn("DROP TABLE", sql)
|
||
self.assertEqual(db.tables[assoc.WORLD_REF][0]["role_code"], evil)
|
||
self.assertEqual(len(db.tables[assoc.WORLD_REF]), 1)
|
||
|
||
|
||
class TestApiContract(unittest.TestCase):
|
||
def setUp(self):
|
||
self.db = new_db()
|
||
|
||
def call(self, name, **params):
|
||
params["_db"] = self.db
|
||
return api.call_api(name, params)
|
||
|
||
def test_upsert_and_list_ok(self):
|
||
res = self.call("pbl_world_ref_upsert", tenant_id=TENANT, world_id=101, class_id=7, team_id=9)
|
||
self.assertTrue(res["ok"], res)
|
||
self.assertEqual(res["data"]["action"], "created")
|
||
page = self.call("pbl_world_ref_list", tenant_id=TENANT, team_id=9)
|
||
self.assertTrue(page["ok"])
|
||
self.assertEqual(page["data"]["total"], 1)
|
||
|
||
def test_scene_and_entity_upsert_ok(self):
|
||
scene = self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9)
|
||
entity = self.call("pbl_entity_ref_upsert", tenant_id=TENANT, entity_id=303, scene_id=202,
|
||
world_id=101, team_id=9, ext={"hp": 100})
|
||
self.assertTrue(scene["ok"], scene)
|
||
self.assertTrue(entity["ok"], entity)
|
||
self.assertEqual(entity["data"]["visibility"], "team")
|
||
|
||
def test_missing_tenant_returns_error_code(self):
|
||
res = self.call("pbl_world_ref_list", world_id=101)
|
||
self.assertFalse(res["ok"])
|
||
self.assertEqual(res["code"], assoc.ERR_TENANT_REQUIRED)
|
||
|
||
def test_unknown_api(self):
|
||
res = api.call_api("pbl_not_exist", {"tenant_id": TENANT})
|
||
self.assertFalse(res["ok"])
|
||
self.assertEqual(res["code"], assoc.ERR_PARAM_REQUIRED)
|
||
|
||
def test_unknown_table_alias(self):
|
||
res = self.call("pbl_domain_ref_get", table="payroll", tenant_id=TENANT, id_value=1)
|
||
self.assertFalse(res["ok"])
|
||
|
||
def test_ref_get_and_delete(self):
|
||
created = self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9)
|
||
rid = created["data"]["id"]
|
||
got = self.call("pbl_domain_ref_get", table="scene", tenant_id=TENANT, scene_id=202, team_id=9)
|
||
self.assertTrue(got["ok"])
|
||
self.assertTrue(got["found"])
|
||
deleted = self.call("pbl_domain_ref_delete", table="pbl_scene_ref", tenant_id=TENANT, id=rid)
|
||
self.assertEqual(deleted["data"]["deleted"], 1)
|
||
missing = self.call("pbl_domain_ref_get", table="scene", tenant_id=TENANT, scene_id=202,
|
||
team_id=9, strict=False)
|
||
self.assertTrue(missing["ok"])
|
||
self.assertFalse(missing["found"])
|
||
|
||
def test_ref_get_strict_not_found(self):
|
||
res = self.call("pbl_domain_ref_get", table="world", tenant_id=TENANT, world_id=777)
|
||
self.assertFalse(res["ok"])
|
||
self.assertEqual(res["code"], assoc.ERR_NOT_FOUND)
|
||
|
||
def test_scope_resolve_and_stats(self):
|
||
self.call("pbl_world_ref_upsert", tenant_id=TENANT, world_id=101, class_id=7, team_id=9)
|
||
self.call("pbl_scene_ref_upsert", tenant_id=TENANT, scene_id=202, world_id=101, team_id=9)
|
||
self.call("pbl_entity_ref_upsert", tenant_id=TENANT, entity_id=303, scene_id=202,
|
||
world_id=101, team_id=9)
|
||
scope = self.call("pbl_domain_scope_resolve", tenant_id=TENANT, entity_id=303)
|
||
self.assertTrue(scope["ok"])
|
||
self.assertEqual(scope["data"]["resolved_from"], "entity")
|
||
self.assertEqual(scope["data"]["world_id"], 101)
|
||
stats = self.call("pbl_domain_stats", tenant_id=TENANT)
|
||
self.assertEqual(stats["data"]["total"], 3)
|
||
self.assertEqual(stats["data"][assoc.ENTITY_REF], 1)
|
||
|
||
def test_materialize_contract(self):
|
||
res = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT,
|
||
game_definition=TestMaterialize.DEFINITION, class_id=7, team_id=9)
|
||
self.assertTrue(res["ok"], res)
|
||
self.assertEqual(res["data"]["world_refs"], 1)
|
||
self.assertEqual(res["data"]["entity_refs"], 2)
|
||
bad = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT,
|
||
game_definition="{oops")
|
||
self.assertFalse(bad["ok"])
|
||
self.assertEqual(bad["code"], assoc.ERR_PARAM_REQUIRED)
|
||
|
||
def test_materialize_missing_definition(self):
|
||
res = self.call("pbl_domain_materialize_game_definition", tenant_id=TENANT)
|
||
self.assertFalse(res["ok"])
|
||
self.assertEqual(res["code"], assoc.ERR_PARAM_REQUIRED)
|
||
|
||
|
||
class TestInitRegistration(unittest.TestCase):
|
||
def test_module_metadata(self):
|
||
from pbl_domain_ext import init as init_mod
|
||
self.assertEqual(init_mod.MODULE_NAME, "pbl_domain_ext")
|
||
for table in (assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF):
|
||
self.assertIn(table, init_mod.OWN_TABLES)
|
||
self.assertEqual(init_mod.READONLY_BASE_TABLES, ["world", "scene", "entity"])
|
||
self.assertIn("pbl_domain_scope_resolve", init_mod.API_NAMES)
|
||
|
||
def test_models_and_cruds_loadable(self):
|
||
from pbl_domain_ext import init as init_mod
|
||
models = init_mod.load_models()
|
||
names = {m.get("tblname") for m in models}
|
||
for table in (assoc.WORLD_REF, assoc.SCENE_REF, assoc.ENTITY_REF):
|
||
self.assertIn(table, names)
|
||
self.assertGreaterEqual(len(init_mod.load_cruds()), 3)
|
||
|
||
def test_load_entry_returns_summary_without_runtime(self):
|
||
from pbl_domain_ext import init as init_mod
|
||
summary = init_mod.load_pbl_domain_ext()
|
||
self.assertEqual(summary["module"], "pbl_domain_ext")
|
||
self.assertIsNone(summary["dbname"]) # 无 ServerEnv 时不抛，只记录错误
|
||
self.assertIn("dbname_error", summary)
|
||
|
||
def test_load_entry_registers_apis_on_env(self):
|
||
from pbl_domain_ext import init as init_mod
|
||
|
||
class FakeEnv(object):
|
||
def __init__(self):
|
||
self.apis = {}
|
||
|
||
def register_api(self, name, handler):
|
||
self.apis[name] = handler
|
||
|
||
env = FakeEnv()
|
||
init_mod.load_pbl_domain_ext(env)
|
||
self.assertIn("pbl_world_ref_list", env.apis)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|