#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M1b 扩展字段 + 关联判定测试（真实执行，非 py_compile）。 覆盖 QC 退回意见 #6（7 类子对象泛化契约可验收证据）与 #5（关联表判定/字段映射证据）： - 7 类子对象走同一组函数签名（泛化），逐类断言 CRUD + 扩展值读写 - 扩展字段校验（必填/枚举/数值区间/未声明字段） - 关联判定 4 种状态：resolved / unresolved / stub / external - Q-OPEN-3 守门：基表写 SQL 被 assert_base_table_immutable 拦截 - 租户隔离：跨租户读不到彼此的子对象与关联 运行： python3 modules/pbl_blueprint/tests/test_m1b_ext_ref.py -v """ import os import shutil import sys # --- M1b sys.path bootstrap: modules/ 下各包互为兄弟仓库，需逐个入 path --- _M1B_MOD_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) _M1B_MODULES_DIR = os.path.abspath(os.path.join(_M1B_MOD_ROOT, "..")) _M1B_CANDIDATES = [_M1B_MOD_ROOT, _M1B_MODULES_DIR] try: for _d in sorted(os.listdir(_M1B_MODULES_DIR)): _sub = os.path.join(_M1B_MODULES_DIR, _d) if os.path.isdir(_sub) and not _d.startswith("."): _M1B_CANDIDATES.append(_sub) except OSError: pass for _p in _M1B_CANDIDATES: if _p not in sys.path: sys.path.insert(0, _p) # --- end bootstrap --- import tempfile import unittest HERE = os.path.dirname(os.path.abspath(__file__)) MOD_ROOT = os.path.dirname(HERE) REPO_ROOT = os.path.abspath(os.path.join(MOD_ROOT, "..", "..")) for p in (REPO_ROOT, MOD_ROOT): if p not in sys.path: sys.path.insert(0, p) import pbl_blueprint.m1b.init as m1b_init # noqa: E402 from pbl_blueprint.m1b.dbutil import ( # noqa: E402 get_conn, reset_conn, sql_exec, sql_rows, table_exists, ) from pbl_blueprint.m1b.errors import ( # noqa: E402 PblForbidden, PblNotFound, PblValidationError, TenantMissingError, ) from pbl_blueprint.m1b.ref import ( # noqa: E402 assert_base_table_immutable, list_refs, ref_contract, ref_summary, resolve_ref, ) from pbl_blueprint.m1b.subobject import ( # noqa: E402 SUBOBJECT_SPECS, create_subobject, delete_subobject, get_ext, get_subobject, list_subobjects, set_ext, subobject_contract, subobject_tree, update_subobject, validate_ext, ) from pbl_blueprint.m1b.tables import REF_KINDS, SUBOBJECT_TYPES # noqa: E402 TENANT = "T_M1B_EXTREF_001" OTHER = "T_M1B_EXTREF_002" BP_ID = "BP_M1B_EXTREF_0001" #: 每类子对象的最小合法载荷（证明 7 类都能通过同一契约创建） SAMPLES = { "driving_question": {"question": "如何让校园垃圾分类真正落地？", "sort_no": 1}, "learning_goal": {"goal": "能设计并执行一次校园垃圾调研", "sort_no": 1}, "mission": {"name": "M1 现状调研", "sort_no": 1}, "problem": {"title": "分类标识不清导致误投", "sort_no": 1}, "role": {"name": "调研组长", "sort_no": 1}, "learner": {"name": "张三", "sort_no": 1}, "artifact_def": {"name": "调研报告", "sort_no": 1}, } #: 每类子对象的扩展字段值（对应 EXT_FIELD_SEED 中声明的字段键） EXT_SAMPLES = { "driving_question": {"cognitive_level": "分析", "weight": 0.5}, "learning_goal": {"bloom_level": "L4", "assessable": True}, "mission": {"estimated_minutes": 60, "unlock_condition": "完成 M0"}, "problem": {"difficulty": "中", "hint_text": "先统计误投率"}, "role": {"team_size": 4, "responsibility": "统筹调研"}, "learner": {"grade": "高一", "group_no": 2}, "artifact_def": {"artifact_type": "文档", "rubric_ref": "rubric_report_v1"}, } class M1bExtRefTestCase(unittest.TestCase): """扩展字段 + 关联判定真实库测试。""" @classmethod def setUpClass(cls): cls.tmpdir = tempfile.mkdtemp(prefix="m1b_extref_") cls.db_path = os.path.join(cls.tmpdir, "pbl_m1b_extref.sqlite3") os.environ["PBL_M1B_DB"] = cls.db_path os.environ["PBL_AUDIT_DIR"] = os.path.join(cls.tmpdir, "audit") reset_conn() cls.conn = get_conn(cls.db_path) m1b_init.load_m1b(conn=cls.conn, actor_id="test_m1b_extref") # 建 7 类子对象基表（最小列集），使泛化 CRUD 可真实落库 for st, spec in SUBOBJECT_SPECS.items(): # 列集去重：text_field 可能就叫 name（如 goal.name）， # 直接字符串拼接会产生 "duplicate column name: name"（sqlite 建表失败） cols, seen = [], set() for col in ("id TEXT PRIMARY KEY", "tenant_key TEXT", "tenant_id TEXT", "blueprint_id TEXT", "%s TEXT" % spec["text_field"], "sort_no INTEGER", "name TEXT", "status TEXT", "created_at TEXT", "updated_at TEXT"): key = col.split()[0].strip("`").lower() if key in seen: continue seen.add(key) cols.append(col) sql_exec( "CREATE TABLE IF NOT EXISTS %s (%s)" % (spec["table"], ", ".join(cols)), conn=cls.conn) sql_exec("CREATE TABLE IF NOT EXISTS pbl_blueprint (" "id TEXT PRIMARY KEY, tenant_key TEXT, tenant_id TEXT, " "name TEXT, code TEXT, status TEXT, template_id TEXT, " "template_code TEXT, template_version TEXT, category TEXT, " "created_at TEXT, updated_at TEXT)", conn=cls.conn) @classmethod def tearDownClass(cls): reset_conn() shutil.rmtree(cls.tmpdir, ignore_errors=True) os.environ.pop("PBL_M1B_DB", None) os.environ.pop("PBL_AUDIT_DIR", None) # ---------- 1. 泛化契约（QC #6） ---------- def test_01_contract_covers_exactly_seven_types(self): """契约声明恰好 7 类，且每类都有表/主键/必填/文本字段声明。""" c = subobject_contract() self.assertEqual(c["type_count"], 7) self.assertEqual(sorted(c["types"]), sorted(SUBOBJECT_TYPES)) self.assertEqual(len(SUBOBJECT_SPECS), 7) for st, spec in SUBOBJECT_SPECS.items(): self.assertTrue(spec["table"].startswith("pbl_")) self.assertEqual(spec["pk"], "id") self.assertIn("blueprint_id", spec["required"]) self.assertTrue(spec["text_field"]) # 9 个统一方法签名对 7 类一致（泛化） self.assertEqual(len(c["unified_methods"]), 9) def test_02_unknown_subobject_type_rejected(self): """非法类型 fail-closed（不允许静默通过）。""" for bad in ("pbl_project", "unknown", "", None): with self.assertRaises(PblValidationError): create_subobject(TENANT, bad, BP_ID, data={"name": "x"}, conn=self.conn) def test_03_all_seven_types_crud_through_same_api(self): """7 类子对象全部通过同一组函数完成 增/查/改/删（泛化实证）。""" created = {} for st in SUBOBJECT_TYPES: row = create_subobject(TENANT, st, BP_ID, data=dict(SAMPLES[st]), actor_id="tester", conn=self.conn) self.assertEqual(row["tenant_id"], TENANT) self.assertEqual(row["blueprint_id"], BP_ID) self.assertTrue(row["id"]) created[st] = row["id"] # 读 got = get_subobject(TENANT, st, row["id"], conn=self.conn) self.assertEqual(got["id"], row["id"]) listed = list_subobjects(TENANT, BP_ID, st, conn=self.conn) self.assertGreaterEqual(len(listed), 1) # 改 upd = update_subobject(TENANT, st, row["id"], data={"sort_no": 9}, conn=self.conn) self.assertEqual(int(upd["sort_no"]), 9) self.assertEqual(len(created), 7) # 删（并级联清理扩展值） for st, oid in created.items(): n = delete_subobject(TENANT, st, oid, conn=self.conn) self.assertEqual(int(n), 1) with self.assertRaises(PblNotFound): get_subobject(TENANT, st, oid, conn=self.conn) def test_04_missing_required_field_rejected(self): """必填字段缺失 -> PblValidationError，且不落库。""" before = sql_rows("SELECT COUNT(*) AS n FROM pbl_mission", conn=self.conn) with self.assertRaises(PblValidationError): create_subobject(TENANT, "mission", BP_ID, data={"sort_no": 1}, conn=self.conn) after = sql_rows("SELECT COUNT(*) AS n FROM pbl_mission", conn=self.conn) self.assertEqual(before[0]["n"], after[0]["n"]) def test_05_tenant_isolation_on_subobjects(self): """跨租户读不到彼此子对象（租户打头强制隔离）。""" row = create_subobject(TENANT, "role", BP_ID, data=dict(SAMPLES["role"]), conn=self.conn) with self.assertRaises(PblNotFound): get_subobject(OTHER, "role", row["id"], conn=self.conn) self.assertEqual(list_subobjects(OTHER, BP_ID, "role", conn=self.conn), []) with self.assertRaises(TenantMissingError): list_subobjects(None, BP_ID, "role", conn=self.conn) # ---------- 2. 扩展字段（QC #6 核心） ---------- def test_06_set_and_get_ext_for_all_seven_types(self): """7 类子对象都能写/读扩展字段值（泛化扩展接口实证）。""" for st in SUBOBJECT_TYPES: row = create_subobject(TENANT, st, BP_ID, data=dict(SAMPLES[st]), conn=self.conn) res = set_ext(TENANT, st, row["id"], EXT_SAMPLES[st], source="manual", blueprint_id=BP_ID, conn=self.conn) self.assertEqual(res["written"], len(EXT_SAMPLES[st])) self.assertEqual(res["problems"], []) got = get_ext(TENANT, st, row["id"], conn=self.conn) for k, v in EXT_SAMPLES[st].items(): self.assertEqual(got.get(k), v, "%s.%s 扩展值不一致" % (st, k)) def test_07_set_ext_is_idempotent(self): """重复写同一扩展字段不产生重复行（唯一键幂等）。""" row = create_subobject(TENANT, "problem", BP_ID, data=dict(SAMPLES["problem"]), conn=self.conn) set_ext(TENANT, "problem", row["id"], {"difficulty": "易"}, conn=self.conn) set_ext(TENANT, "problem", row["id"], {"difficulty": "难"}, conn=self.conn) rows = sql_rows( "SELECT COUNT(*) AS n FROM pbl_subobject_ext WHERE subobject_id = ? " "AND field_key = 'difficulty'", [row["id"]], conn=self.conn) self.assertEqual(int(rows[0]["n"]), 1) self.assertEqual(get_ext(TENANT, "problem", row["id"], conn=self.conn)["difficulty"], "难") def test_08_ext_validation_enum_and_range(self): """枚举越界 / 数值越界 / 类型错误 / 未声明字段 均被校验拦截。""" # 枚举越界 p = validate_ext(TENANT, "driving_question", {"cognitive_level": "不存在级"}, conn=self.conn) self.assertTrue(any(x["code"] == "ENUM" for x in p), p) # 数值越界（weight max=1） p = validate_ext(TENANT, "driving_question", {"weight": 5}, conn=self.conn) self.assertTrue(any(x["code"] == "MAX" for x in p), p) # 类型错误 p = validate_ext(TENANT, "mission", {"estimated_minutes": "很多"}, conn=self.conn) self.assertTrue(any(x["code"] == "TYPE" for x in p), p) # 未声明字段 p = validate_ext(TENANT, "role", {"not_declared_field": 1}, conn=self.conn) self.assertTrue(any(x["code"] == "UNDECLARED" for x in p), p) # 合法值通过 self.assertEqual(validate_ext(TENANT, "learner", {"grade": "高二", "group_no": 3}, conn=self.conn), []) def test_09_set_ext_rejects_invalid_values(self): """set_ext 校验失败即抛错且不落库。""" row = create_subobject(TENANT, "artifact_def", BP_ID, data=dict(SAMPLES["artifact_def"]), conn=self.conn) before = sql_rows("SELECT COUNT(*) AS n FROM pbl_subobject_ext " "WHERE subobject_id = ?", [row["id"]], conn=self.conn) with self.assertRaises(PblValidationError): set_ext(TENANT, "artifact_def", row["id"], {"artifact_type": "不存在的类型"}, conn=self.conn) after = sql_rows("SELECT COUNT(*) AS n FROM pbl_subobject_ext " "WHERE subobject_id = ?", [row["id"]], conn=self.conn) self.assertEqual(before[0]["n"], after[0]["n"]) def test_10_create_with_inline_ext(self): """创建子对象时可内联 ext，一次调用完成基础行+扩展值。""" row = create_subobject( TENANT, "learning_goal", BP_ID, data=dict(SAMPLES["learning_goal"], ext={"bloom_level": "L5", "assessable": True}), conn=self.conn) self.assertEqual(row["ext"]["bloom_level"], "L5") self.assertEqual(get_ext(TENANT, "learning_goal", row["id"], conn=self.conn)["assessable"], True) def test_11_delete_cascades_ext_values(self): """删除子对象级联清理扩展值（应用层级联，无外键）。""" row = create_subobject(TENANT, "mission", BP_ID, data=dict(SAMPLES["mission"]), conn=self.conn) set_ext(TENANT, "mission", row["id"], {"estimated_minutes": 30}, conn=self.conn) self.assertGreater(len(get_ext(TENANT, "mission", row["id"], conn=self.conn)), 0) delete_subobject(TENANT, "mission", row["id"], conn=self.conn) left = sql_rows("SELECT COUNT(*) AS n FROM pbl_subobject_ext " "WHERE subobject_id = ?", [row["id"]], conn=self.conn) self.assertEqual(int(left[0]["n"]), 0) def test_12_subobject_tree_aggregates_seven_types(self): """subobject_tree 一次取全 7 类 + 扩展值。""" bp = "BP_TREE_%s" % os.getpid() for st in SUBOBJECT_TYPES: create_subobject(TENANT, st, bp, data=dict(SAMPLES[st]), conn=self.conn) oid = list_subobjects(TENANT, bp, st, conn=self.conn)[0]["id"] set_ext(TENANT, st, oid, EXT_SAMPLES[st], conn=self.conn, blueprint_id=bp) tree = subobject_tree(TENANT, bp, conn=self.conn) self.assertEqual(sorted(tree["subobjects"].keys()), sorted(SUBOBJECT_TYPES)) self.assertEqual(tree["counts"]["__total__"], 7) for st in SUBOBJECT_TYPES: self.assertEqual(len(tree["subobjects"][st]), 1) self.assertTrue(tree["subobjects"][st][0]["ext"]) # ---------- 3. 关联表判定（QC #5 + Q-OPEN-3） ---------- def test_13_ref_contract_declares_readonly_and_no_fk(self): """关联契约声明：4 域只读、无外键、基表零写入。""" c = ref_contract() self.assertEqual(sorted(c["ref_kinds"]), sorted(REF_KINDS)) self.assertEqual(c["ownership"], "read_only") self.assertTrue(c["q_open_3"]["no_foreign_key"]) self.assertEqual(c["q_open_3"]["base_table_writes"], 0) self.assertEqual(sorted(c["base_tables_readonly"].keys()), ["entity", "scene", "script", "world"]) def test_14_ref_stub_when_base_table_unavailable(self): """基表不可达 -> resolve_status='stub'，不报错、不建表（Q-OPEN-3）。""" row = resolve_ref(TENANT, BP_ID, "world", "W_NOT_EXIST_1", actor_id="tester", conn=self.conn) self.assertEqual(row["resolve_status"], "stub") self.assertEqual(row["ownership"], "read_only") self.assertIn("not available", row["resolve_note"]) self.assertFalse(table_exists("world", conn=self.conn), "判定过程不得创建 world 基表") def test_15_ref_resolved_with_readonly_snapshot(self): """基表可达且行存在 -> resolved + 只读快照。""" sql_exec("CREATE TABLE IF NOT EXISTS world (id TEXT PRIMARY KEY, " "name TEXT, code TEXT, status TEXT)", conn=self.conn) sql_exec("INSERT INTO world (id, name, code, status) VALUES (?,?,?,?)", ["W_M1B_1", "M1b 测试世界", "world_m1b_1", "published"], conn=self.conn) row = resolve_ref(TENANT, BP_ID, "world", "W_M1B_1", conn=self.conn) self.assertEqual(row["resolve_status"], "resolved") self.assertEqual(row["ref_table"], "world") self.assertEqual(row["ref_code"], "world_m1b_1") self.assertTrue(row["ref_snapshot_json"]) # 快照是只读副本：改基表不影响已落快照 sql_exec("UPDATE world SET name = '改名后' WHERE id = 'W_M1B_1'", conn=self.conn) stored = [r for r in list_refs(TENANT, blueprint_id=BP_ID, conn=self.conn) if r["ref_id"] == "W_M1B_1"][0] self.assertEqual(stored["ref_snapshot"]["name"], "M1b 测试世界") def test_16_ref_unresolved_when_row_missing(self): """基表可达但行不存在 -> unresolved。""" row = resolve_ref(TENANT, BP_ID, "world", "W_MISSING_ROW", conn=self.conn) self.assertEqual(row["resolve_status"], "unresolved") self.assertIn("not found", row["resolve_note"]) def test_17_ref_external_is_stub(self): """external 域恒为 stub（M1b 不解析外部系统）。""" row = resolve_ref(TENANT, BP_ID, "external", "EXT_1", conn=self.conn) self.assertEqual(row["resolve_status"], "stub") self.assertEqual(row["ref_table"], "") def test_18_ref_invalid_kind_rejected(self): """非法 ref_kind / 缺 ref_id -> PblValidationError。""" with self.assertRaises(PblValidationError): resolve_ref(TENANT, BP_ID, "not_a_kind", "X1", conn=self.conn) with self.assertRaises(PblValidationError): resolve_ref(TENANT, BP_ID, "world", "", conn=self.conn) with self.assertRaises(TenantMissingError): resolve_ref(None, BP_ID, "world", "X1", conn=self.conn) def test_19_ref_idempotent_and_subobject_scoped(self): """同目标+同发起方重复判定不增行；子对象级关联可区分。""" r1 = resolve_ref(TENANT, BP_ID, "scene", "S_M1B_1", conn=self.conn) r2 = resolve_ref(TENANT, BP_ID, "scene", "S_M1B_1", conn=self.conn) self.assertEqual(r1["id"], r2["id"]) n = sql_rows("SELECT COUNT(*) AS c FROM pbl_blueprint_ref " "WHERE ref_id = 'S_M1B_1' AND subobject_type = 'blueprint'", conn=self.conn) self.assertEqual(int(n[0]["c"]), 1) row = create_subobject(TENANT, "mission", BP_ID, data=dict(SAMPLES["mission"]), conn=self.conn) resolve_ref(TENANT, BP_ID, "scene", "S_M1B_1", subobject_type="mission", subobject_id=row["id"], conn=self.conn) n2 = sql_rows("SELECT COUNT(*) AS c FROM pbl_blueprint_ref " "WHERE ref_id = 'S_M1B_1'", conn=self.conn) self.assertEqual(int(n2[0]["c"]), 2, "子对象级关联应独立成行") def test_20_ref_summary_and_tenant_isolation(self): """汇总报表按 kind×status 计数；跨租户不可见。""" resolve_ref(OTHER, BP_ID, "entity", "E_OTHER_1", conn=self.conn) s = ref_summary(TENANT, blueprint_id=BP_ID, conn=self.conn) self.assertEqual(s["ownership"], "read_only") self.assertEqual(s["base_tables_modified"], 0) self.assertGreaterEqual(s["total"], 3) self.assertIn("world", s["by_kind"]) self.assertEqual(s["by_status"].get("resolved", 0), 1) rows = list_refs(OTHER, blueprint_id=BP_ID, conn=self.conn) self.assertTrue(all(r["tenant_id"] == OTHER for r in rows)) self.assertFalse(any(r["ref_id"] == "E_OTHER_1" for r in list_refs(TENANT, conn=self.conn))) def test_21_base_table_write_sql_is_blocked(self): """Q-OPEN-3 守门：对基表的写 SQL 被静态拦截。""" bad = [ "UPDATE world SET name = 'x' WHERE id = '1'", "DELETE FROM scene WHERE id = '1'", "INSERT INTO entity (id) VALUES ('1')", "ALTER TABLE script ADD COLUMN tenant_id VARCHAR(64)", ] for sql in bad: with self.assertRaises(PblForbidden): assert_base_table_immutable(sql) # 只读与自有表写入放行 self.assertTrue(assert_base_table_immutable( "SELECT * FROM world WHERE id = '1'")) self.assertTrue(assert_base_table_immutable( "INSERT INTO pbl_blueprint_ref (id) VALUES ('1')")) def test_22_generated_ddl_contains_no_fk_and_no_base_table(self): """生成的 sqlite/mariadb DDL 均无 FOREIGN KEY、不含基表建表语句。""" ddl = m1b_init.SQLITE_DDL.upper() self.assertNotIn("FOREIGN KEY", ddl) for base in ("CREATE TABLE IF NOT EXISTS WORLD", "CREATE TABLE IF NOT EXISTS SCENE", "CREATE TABLE IF NOT EXISTS ENTITY", "CREATE TABLE IF NOT EXISTS SCRIPT"): self.assertNotIn(base, ddl) for t in TABLES: self.assertIn("CREATE TABLE IF NOT EXISTS %s" % t["name"].upper(), ddl) if __name__ == "__main__": unittest.main(verbosity=2)