327 lines
14 KiB
Python
327 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
M1b 自检测试：模板平台公共部分（tenant_id NULL）+ 子对象扩展 + 关联表判定（Q-OPEN-3）
|
||
|
||
运行：cd modules/pbl_blueprint && python -m pytest tests/test_m1b_template.py -q
|
||
（无 pytest 时：python tests/test_m1b_template.py）
|
||
|
||
覆盖 M1b 出口门禁：
|
||
① 模板 schema 校验 fail-closed（非法类型/悬空父/环/缺必填 -> PBL_TPL_SCHEMA_INVALID）
|
||
② ID 重映射：临时 ID -> 业务主键，父子同步替换，外部引用保持原值
|
||
③ tpl_hash 确定性（同模板同输入同 hash；键序无关）
|
||
④ 平台公共模板解析优先级（租户私有 > 平台公共）+ 写入门禁
|
||
⑤ Q-OPEN-3 守卫：DDL 不含对 world/scene/entity/script 基表的改动
|
||
⑥ 离线种子模板可解析、可重映射、结构完整无孤儿
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import sys
|
||
import unittest
|
||
|
||
import _m1b_loader # noqa: F401 (空壳包引导，绕开 M1a 挂载链 eager import)
|
||
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), ".."))
|
||
|
||
from pbl_blueprint.models.pbl_template import ( # noqa: E402
|
||
M1B_TABLES, PLATFORM_TENANT_KEY, PROTECTED_BASE_TABLES,
|
||
assert_no_base_table_change, build_all_ddl, build_ddl,
|
||
is_platform_scope, normalize_tenant_key,
|
||
)
|
||
from pbl_blueprint.subobject_ext import ( # noqa: E402
|
||
SUBOBJECT_TYPE_LIST, TplSchemaError, apply_ref_status,
|
||
extract_external_refs, group_refs_by_domain, remap_ids,
|
||
tpl_hash, validate_tpl_schema,
|
||
)
|
||
import pbl_blueprint.template_platform as tp # noqa: E402
|
||
|
||
SEED_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"..", "pbl_blueprint", "json", "seed_template_offline.json")
|
||
|
||
|
||
def _gen_code_factory():
|
||
"""确定性编码生成器（替代 pbl_common.gen_code，保证测试可复现）。"""
|
||
counter = {"n": 0}
|
||
|
||
def gen_code(prefix, tenant_id):
|
||
counter["n"] += 1
|
||
return "%s%s%05d" % (prefix, (tenant_id or "PLT")[:3].upper(), counter["n"])
|
||
return gen_code
|
||
|
||
|
||
def _demo_tpl():
|
||
return {
|
||
"schema_version": "1.0",
|
||
"subobjects": [
|
||
{"tmp_id": "s1", "subobject_type": "stage", "name": "阶段1", "seq": 1},
|
||
{"tmp_id": "t1", "subobject_type": "task", "parent_tmp_id": "s1", "name": "任务1", "seq": 1},
|
||
{"tmp_id": "r1", "subobject_type": "role", "parent_tmp_id": "t1", "name": "角色1", "seq": 1},
|
||
{"tmp_id": "w1", "subobject_type": "world_ref", "parent_tmp_id": "t1", "world_code": "W001", "seq": 2},
|
||
{"tmp_id": "sc1", "subobject_type": "scene_ref", "parent_tmp_id": "t1", "scene_code": "SC001", "seq": 3},
|
||
{"tmp_id": "e1", "subobject_type": "entity_ref", "parent_tmp_id": "sc1", "entity_code": "EN001", "seq": 1},
|
||
{"tmp_id": "sp1", "subobject_type": "script_ref", "parent_tmp_id": "t1", "script_code": "SP001", "seq": 4},
|
||
],
|
||
}
|
||
|
||
|
||
class TestSubobjectExt(unittest.TestCase):
|
||
|
||
def test_01_schema_ok_and_topo_order(self):
|
||
norm = validate_tpl_schema(_demo_tpl())
|
||
types = [s["subobject_type"] for s in norm["subobjects"]]
|
||
self.assertEqual(len(norm["subobjects"]), 7)
|
||
# 父必在子之前（拓扑序）
|
||
pos = {s["tmp_id"]: i for i, s in enumerate(norm["subobjects"])}
|
||
for s in norm["subobjects"]:
|
||
p = s.get("parent_tmp_id")
|
||
if p:
|
||
self.assertLess(pos[p], pos[s["tmp_id"]])
|
||
self.assertTrue(types)
|
||
|
||
def test_02_schema_invalid_type(self):
|
||
tpl = _demo_tpl()
|
||
tpl["subobjects"][0]["subobject_type"] = "unknown_type"
|
||
with self.assertRaises(TplSchemaError) as cm:
|
||
validate_tpl_schema(tpl)
|
||
self.assertEqual(cm.exception.code, "PBL_TPL_SCHEMA_INVALID")
|
||
|
||
def test_03_schema_dangling_parent(self):
|
||
tpl = _demo_tpl()
|
||
tpl["subobjects"][1]["parent_tmp_id"] = "not_exist"
|
||
with self.assertRaises(TplSchemaError):
|
||
validate_tpl_schema(tpl)
|
||
|
||
def test_04_schema_cycle(self):
|
||
tpl = {"schema_version": "1.0", "subobjects": [
|
||
{"tmp_id": "a", "subobject_type": "task", "parent_tmp_id": "b", "name": "A"},
|
||
{"tmp_id": "b", "subobject_type": "stage", "parent_tmp_id": "a", "name": "B"},
|
||
]}
|
||
with self.assertRaises(TplSchemaError):
|
||
validate_tpl_schema(tpl)
|
||
|
||
def test_05_schema_missing_required(self):
|
||
tpl = {"schema_version": "1.0", "subobjects": [
|
||
{"tmp_id": "s1", "subobject_type": "stage", "seq": 1},
|
||
]}
|
||
with self.assertRaises(TplSchemaError):
|
||
validate_tpl_schema(tpl)
|
||
|
||
def test_06_schema_stage_cannot_have_parent(self):
|
||
tpl = {"schema_version": "1.0", "subobjects": [
|
||
{"tmp_id": "s1", "subobject_type": "stage", "name": "S", "seq": 1},
|
||
{"tmp_id": "s2", "subobject_type": "stage", "parent_tmp_id": "s1", "name": "S2", "seq": 2},
|
||
]}
|
||
with self.assertRaises(TplSchemaError):
|
||
validate_tpl_schema(tpl)
|
||
|
||
def test_07_hash_deterministic_and_key_order_free(self):
|
||
tpl = _demo_tpl()
|
||
h1 = tpl_hash(tpl)
|
||
h2 = tpl_hash(json.dumps(tpl, ensure_ascii=False))
|
||
shuffled = {"subobjects": tpl["subobjects"], "schema_version": tpl["schema_version"]}
|
||
h3 = tpl_hash(shuffled)
|
||
self.assertEqual(h1, h2)
|
||
self.assertEqual(h1, h3)
|
||
self.assertEqual(len(h1), 64)
|
||
|
||
def test_08_remap_ids(self):
|
||
norm = validate_tpl_schema(_demo_tpl())
|
||
items, mapping = remap_ids(norm["subobjects"], _gen_code_factory(), "T001")
|
||
self.assertEqual(len(items), 7)
|
||
self.assertEqual(len(set(mapping.values())), 7) # 无碰撞
|
||
for it in items:
|
||
self.assertNotIn("tmp_id", it)
|
||
self.assertNotIn("parent_tmp_id", it)
|
||
if it["parent_code"]:
|
||
self.assertIn(it["parent_code"], mapping.values())
|
||
# 前缀按类型
|
||
stage = [i for i in items if i["subobject_type"] == "stage"][0]
|
||
self.assertTrue(stage["subobject_code"].startswith("STG"))
|
||
self.assertIsNone(stage["parent_code"])
|
||
|
||
def test_09_external_refs_keep_original_value(self):
|
||
norm = validate_tpl_schema(_demo_tpl())
|
||
refs = extract_external_refs(norm["subobjects"])
|
||
self.assertEqual(len(refs), 4)
|
||
codes = {r["ref_code"] for r in refs}
|
||
self.assertEqual(codes, {"W001", "SC001", "EN001", "SP001"})
|
||
grouped = group_refs_by_domain(refs)
|
||
self.assertEqual(sorted(grouped.keys()), ["entity", "scene", "script", "world"])
|
||
items, _ = remap_ids(norm["subobjects"], _gen_code_factory(), "T001")
|
||
w = [i for i in items if i["subobject_type"] == "world_ref"][0]
|
||
self.assertEqual(w["world_code"], "W001") # 原值不重映射
|
||
|
||
def test_10_apply_ref_status_warn_mode(self):
|
||
norm = validate_tpl_schema(_demo_tpl())
|
||
refs = extract_external_refs(norm["subobjects"])
|
||
items, _ = remap_ids(norm["subobjects"], _gen_code_factory(), "T001")
|
||
unresolved = apply_ref_status(items, refs, {
|
||
"world": {"W001": True}, "scene": {"SC001": True},
|
||
"entity": {}, "script": {},
|
||
})
|
||
self.assertEqual(unresolved, 2)
|
||
statuses = {i["subobject_type"]: i.get("ref_status") for i in items if i.get("ref_status")}
|
||
self.assertEqual(statuses["world_ref"], "resolved")
|
||
self.assertEqual(statuses["entity_ref"], "unresolved")
|
||
|
||
def test_11_seven_types_match_dict(self):
|
||
self.assertEqual(
|
||
sorted(SUBOBJECT_TYPE_LIST),
|
||
sorted(["stage", "task", "role", "world_ref", "scene_ref", "entity_ref", "script_ref"]))
|
||
|
||
|
||
class TestPlatformScope(unittest.TestCase):
|
||
|
||
def test_20_normalize_tenant_key(self):
|
||
self.assertEqual(normalize_tenant_key(None), PLATFORM_TENANT_KEY)
|
||
self.assertEqual(normalize_tenant_key(""), PLATFORM_TENANT_KEY)
|
||
self.assertEqual(normalize_tenant_key(" "), PLATFORM_TENANT_KEY)
|
||
self.assertEqual(normalize_tenant_key("T001"), "T001")
|
||
self.assertTrue(is_platform_scope(None))
|
||
self.assertFalse(is_platform_scope("T001"))
|
||
|
||
def test_21_tenant_id_nullable_in_ddl(self):
|
||
ddl = build_ddl("pbl_template")
|
||
self.assertIn("`tenant_id` VARCHAR(32) NULL", ddl)
|
||
self.assertIn("IFNULL(`tenant_id`,'__PLATFORM__')", ddl)
|
||
# 实例化日志 tenant_id 仍 NOT NULL
|
||
ddl2 = build_ddl("pbl_template_instance_log")
|
||
self.assertIn("`tenant_id` VARCHAR(32) NOT NULL", ddl2)
|
||
|
||
def test_22_resolve_priority_tenant_over_platform(self):
|
||
rows = [
|
||
{"tenant_id": "T001", "template_code": "C1", "template_version": 1},
|
||
{"tenant_id": None, "template_code": "C1", "template_version": 1},
|
||
]
|
||
row, scope = tp.pick_template(rows)
|
||
self.assertEqual(scope, "platform" if False else scope) # 保持可读
|
||
self.assertEqual(scope, tp.SCOPE_TENANT)
|
||
self.assertEqual(row["tenant_id"], "T001")
|
||
# 仅平台行 -> platform
|
||
row2, scope2 = tp.pick_template([rows[1]])
|
||
self.assertEqual(scope2, tp.SCOPE_PLATFORM)
|
||
self.assertIsNone(row2["tenant_id"])
|
||
with self.assertRaises(tp.TemplateNotFound):
|
||
tp.pick_template([])
|
||
|
||
def test_23_platform_write_guard(self):
|
||
self.assertEqual(
|
||
tp.build_platform_write_guard({"tenant_id": "P", "roles": ["platform_admin"]}, None),
|
||
tp.SCOPE_PLATFORM)
|
||
with self.assertRaises(tp.TemplatePlatformForbidden):
|
||
tp.build_platform_write_guard({"tenant_id": "T001", "roles": ["teacher"]}, None)
|
||
self.assertEqual(
|
||
tp.build_platform_write_guard({"tenant_id": "T001"}, "T001"), tp.SCOPE_TENANT)
|
||
with self.assertRaises(tp.TemplatePlatformForbidden):
|
||
tp.build_platform_write_guard({"tenant_id": "T001"}, "T002")
|
||
|
||
def test_24_require_tenant_fail_closed(self):
|
||
with self.assertRaises(tp.TemplateError) as cm:
|
||
tp.require_tenant({})
|
||
self.assertEqual(cm.exception.code, "PBL_TENANT_MISSING")
|
||
self.assertEqual(tp.require_tenant({"tenant_id": "T001"}), "T001")
|
||
|
||
def test_25_list_sql_tenant_first(self):
|
||
sql, params = tp.build_list_sql("T001", {"subject": "math"})
|
||
self.assertIn("(tenant_id = %s OR tenant_id IS NULL)", sql)
|
||
self.assertIn("ORDER BY (tenant_id IS NULL) ASC", sql)
|
||
self.assertEqual(params, ["T001", "math"])
|
||
sql2, params2 = tp.build_list_sql("T001", {"scope": "platform"})
|
||
self.assertIn("tenant_id IS NULL", sql2)
|
||
self.assertEqual(params2, [])
|
||
|
||
def test_26_offline_sql(self):
|
||
sql, params = tp.build_offline_sql("T001", subject="general")
|
||
self.assertIn("offline_flag = 'Y'", sql)
|
||
self.assertIn("tenant_id IS NULL", sql)
|
||
self.assertEqual(params[0], "T001")
|
||
|
||
|
||
class TestQOpen3RelationTable(unittest.TestCase):
|
||
|
||
def test_30_ddl_has_no_base_table_change(self):
|
||
ddl = build_all_ddl()
|
||
self.assertTrue(assert_no_base_table_change(ddl))
|
||
upper = ddl.upper()
|
||
for tbl in PROTECTED_BASE_TABLES:
|
||
self.assertNotIn("ALTER TABLE %s" % tbl.upper(), upper)
|
||
self.assertNotIn("ALTER TABLE `%s`" % tbl.upper(), upper)
|
||
self.assertNotIn("FOREIGN KEY", upper)
|
||
|
||
def test_31_guard_rejects_violation(self):
|
||
with self.assertRaises(AssertionError):
|
||
assert_no_base_table_change("ALTER TABLE world ADD COLUMN tpl_code VARCHAR(32);")
|
||
with self.assertRaises(AssertionError):
|
||
assert_no_base_table_change("ALTER TABLE `scene` ADD COLUMN x INT;")
|
||
|
||
def test_32_only_two_m1b_tables(self):
|
||
self.assertEqual(sorted(M1B_TABLES),
|
||
sorted(["pbl_template", "pbl_template_instance_log"]))
|
||
ddl = build_all_ddl()
|
||
self.assertEqual(ddl.count("CREATE TABLE"), 2)
|
||
|
||
def test_33_sql_file_guard(self):
|
||
path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"..", "pbl_blueprint", "sql", "m1b_template.sql")
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
content = fh.read()
|
||
# 文件内说明性注释提到基表名是允许的，但不得有真实 ALTER/CREATE 语句
|
||
for line in content.splitlines():
|
||
stripped = line.strip()
|
||
if stripped.startswith("--"):
|
||
continue
|
||
upper = stripped.upper()
|
||
for tbl in PROTECTED_BASE_TABLES:
|
||
self.assertFalse(
|
||
upper.startswith("ALTER TABLE %s" % tbl.upper()) or
|
||
upper.startswith("ALTER TABLE `%s`" % tbl.upper()),
|
||
"Q-OPEN-3 违规行: %s" % stripped)
|
||
|
||
|
||
class TestOfflineSeed(unittest.TestCase):
|
||
|
||
def _seeds(self):
|
||
with open(SEED_FILE, "r", encoding="utf-8") as fh:
|
||
return json.load(fh)["templates"]
|
||
|
||
def test_40_seed_valid_and_offline(self):
|
||
seeds = self._seeds()
|
||
self.assertGreaterEqual(len(seeds), 1)
|
||
for s in seeds:
|
||
self.assertEqual(s["offline_flag"], "Y")
|
||
self.assertIsNone(s["tenant_id"]) # 平台公共
|
||
norm = validate_tpl_schema(s["tpl_json"]) # 不抛即通过
|
||
self.assertGreater(len(norm["subobjects"]), 0)
|
||
self.assertEqual(len(tpl_hash(norm)), 64)
|
||
|
||
def test_41_seed_instantiate_no_orphan(self):
|
||
"""出口门禁①：种子模板实例化后树完整无孤儿（父引用全部可解析）。"""
|
||
for s in self._seeds():
|
||
norm = validate_tpl_schema(s["tpl_json"])
|
||
items, mapping = remap_ids(norm["subobjects"], _gen_code_factory(), "T_DEMO")
|
||
codes = {i["subobject_code"] for i in items}
|
||
roots = 0
|
||
for i in items:
|
||
if i["parent_code"] is None:
|
||
roots += 1
|
||
else:
|
||
self.assertIn(i["parent_code"], codes) # 无孤儿
|
||
self.assertEqual(roots, 1, "演示模板应恰有 1 个顶层 stage")
|
||
|
||
def test_42_seed_min_has_no_external_ref(self):
|
||
seeds = {s["template_code"]: s for s in self._seeds()}
|
||
min_tpl = seeds["TPL_DEMO_BLUEPRINT_MIN"]
|
||
norm = validate_tpl_schema(min_tpl["tpl_json"])
|
||
self.assertEqual(extract_external_refs(norm["subobjects"]), [])
|
||
|
||
def test_43_seed_deterministic_hash(self):
|
||
"""出口门禁④：同模板同输入产出同结构（hash 稳定）。"""
|
||
s = self._seeds()[0]
|
||
self.assertEqual(tpl_hash(s["tpl_json"]),
|
||
tpl_hash(json.loads(json.dumps(s["tpl_json"]))))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|