# -*- 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)