#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M1b 真实库测试：建表/注册同步/模板平台公共/实例化/离线兜底/幂等。 对应 QC 退回意见 #5/#7/#8/#9：本文件**真实执行**（非 py_compile）， 使用 sqlite 文件库落盘，断言库内行数与字段值，执行日志由 tools/m1b_run_tests.py 落盘到 projects/pbls/deliverables/m1b/test_logs/。 运行： python3 modules/pbl_blueprint/tests/test_m1b_realdb.py -v """ import json 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_rows, sql_scalar, table_exists, ) from pbl_blueprint.m1b.errors import ( # noqa: E402 ErrorCode, PblConflict, PblForbidden, PblNotFound, PblValidationError, TenantMissingError, ) from pbl_blueprint.m1b.tables import SUBOBJECT_TYPES, TABLES # noqa: E402 from pbl_blueprint.m1b.tenant import require_tenant # noqa: E402 TENANT = "T_M1B_REALDB_001" OTHER_TENANT = "T_M1B_REALDB_002" class M1bRealDbTestCase(unittest.TestCase): """真实 sqlite 文件库上的 M1b 端到端断言。""" @classmethod def setUpClass(cls): cls.tmpdir = tempfile.mkdtemp(prefix="m1b_realdb_") cls.db_path = os.path.join(cls.tmpdir, "pbl_m1b_test.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) cls.load_result = m1b_init.load_m1b(conn=cls.conn, actor_id="test_m1b_realdb") @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. DDL 落库证据 ---------- def test_01_all_four_tables_created(self): """4 张 M1b 表全部真实建出（QC #7：表结构落库证据）。""" ready = [t["name"] for t in TABLES if table_exists(t["name"], conn=self.conn)] self.assertEqual(sorted(ready), sorted([t["name"] for t in TABLES]), "建表缺失: %s" % set(t["name"] for t in TABLES) - set(ready)) self.assertEqual(self.load_result["ddl"]["missing"], []) self.assertGreater(self.load_result["ddl"]["statements"], 0) def test_02_ddl_is_idempotent(self): """重复执行 load_m1b 不报错、不产生重复注册行（QC #8：幂等）。""" before = {t["name"]: sql_scalar("SELECT COUNT(*) FROM %s" % t["name"], conn=self.conn) for t in TABLES} again = m1b_init.load_m1b(conn=self.conn, actor_id="test_idempotent") after = {t["name"]: sql_scalar("SELECT COUNT(*) FROM %s" % t["name"], conn=self.conn) for t in TABLES} self.assertEqual(before, after, "重复执行导致行数变化，非幂等") self.assertEqual(again["ext_field_defs"]["created"], 0, "第二次注册不应新增扩展字段定义") self.assertGreater(again["ext_field_defs"]["updated"], 0) def test_03_indexes_exist_and_unique_keys_tenant_first(self): """索引真实建出，且唯一键以 tenant_key 打头（规避 NULL 重复行）。""" for t in TABLES: rows = sql_rows("PRAGMA index_list(%s)" % t["name"], conn=self.conn) names = [r["name"] for r in rows] for ix in t["indexes"]: self.assertIn(ix["name"], names, "%s 缺索引 %s" % (t["name"], ix["name"])) if ix.get("unique"): self.assertEqual(ix["columns"][0], "tenant_key", "%s 唯一键未以 tenant_key 打头" % ix["name"]) def test_04_no_foreign_key_declared(self): """Q-OPEN-3：4 张表均无 FOREIGN KEY。""" for t in TABLES: fks = sql_rows("PRAGMA foreign_key_list(%s)" % t["name"], conn=self.conn) self.assertEqual(fks, [], "%s 不应声明外键" % t["name"]) def test_05_columns_match_table_definition(self): """库内列与表定义字段一一对应（模型-DDL-库三方一致，QC #5）。""" for t in TABLES: rows = sql_rows("PRAGMA table_info(%s)" % t["name"], conn=self.conn) db_cols = [r["name"] for r in rows] expect = [f["name"] for f in t["fields"]] self.assertEqual(sorted(db_cols), sorted(expect), "%s 列与定义不一致" % t["name"]) # ---------- 2. 注册同步证据 ---------- def test_06_ext_field_defs_registered_for_all_seven_types(self): """7 类子对象都注册到扩展字段定义（QC #6/#8：库内注册数据）。""" rows = sql_rows( "SELECT DISTINCT owner_type FROM pbl_ext_field_def " "WHERE tenant_id IS NULL ORDER BY owner_type", conn=self.conn) owners = [r["owner_type"] for r in rows] for st in SUBOBJECT_TYPES: self.assertIn(st, owners, "子对象类型 %s 未注册扩展字段定义" % st) self.assertIn("template", owners) self.assertIn("blueprint", owners) n = sql_scalar("SELECT COUNT(*) FROM pbl_ext_field_def", conn=self.conn) self.assertEqual(int(n), len(m1b_init.EXT_FIELD_SEED)) def test_07_platform_templates_have_null_tenant_id(self): """平台公共模板 tenant_id 为 NULL、tenant_key='__platform__'（QC #5 核心）。""" rows = sql_rows( "SELECT id, code, version, scope, tenant_id, tenant_key, status " "FROM pbl_blueprint_template WHERE tenant_id IS NULL " "ORDER BY code", conn=self.conn) self.assertGreaterEqual(len(rows), 1, "未注册任何平台公共模板") for r in rows: self.assertIsNone(r["tenant_id"]) self.assertEqual(r["tenant_key"], "__platform__") self.assertEqual(r["scope"], "platform") self.assertEqual(r["status"], "published") def test_08_audit_trail_written(self): """注册同步产生审计轨迹（append-only 落盘）。""" path = os.path.join(os.environ["PBL_AUDIT_DIR"], "pbl_blueprint_audit.jsonl") self.assertTrue(os.path.exists(path), "审计 JSONL 未落盘: %s" % path) with open(path, "r", encoding="utf-8") as fh: lines = [ln for ln in fh.read().splitlines() if ln.strip()] self.assertGreaterEqual(len(lines), 1) actions = set() for ln in lines: rec = json.loads(ln) actions.add(rec["action"]) self.assertIn("id", rec) self.assertIn("created_at", rec) self.assertIn("ddl_apply", actions) self.assertIn("register_sync", actions) # ---------- 3. 模板平台公共部分 ---------- def test_09_tenant_can_read_platform_template_but_not_write(self): """租户可读平台公共模板；租户写平台模板被拒（写保护）。""" from pbl_blueprint.m1b.template import ( get_template, list_templates, update_template, ) items = list_templates(tenant_id=TENANT, conn=self.conn) plat = [i for i in items if i.get("is_platform")] self.assertGreaterEqual(len(plat), 1, "租户看不到平台公共模板") tpl = get_template(tenant_id=TENANT, template_id=plat[0]["id"], conn=self.conn, allow_offline=False) self.assertIsNone(tpl["tenant_id"]) self.assertTrue(tpl["is_platform"]) with self.assertRaises(PblForbidden): update_template(tenant_id=TENANT, template_id=tpl["id"], data={"name": "租户篡改"}, conn=self.conn) # 平台角色可写 ok_row = update_template(tenant_id=None, template_id=tpl["id"], data={"description": "平台侧更新"}, role="platform", conn=self.conn) self.assertEqual(ok_row["description"], "平台侧更新") def test_10_tenant_cannot_create_platform_template(self): """非平台角色创建 scope=platform 模板 -> PblForbidden。""" from pbl_blueprint.m1b.template import create_template with self.assertRaises(PblForbidden): create_template(tenant_id=TENANT, data={"code": "hack.platform", "name": "越权", "scope": "platform"}, conn=self.conn) def test_11_tenant_template_requires_tenant_id(self): """租户模板缺 tenant_id -> TenantMissingError（fail-closed）。""" from pbl_blueprint.m1b.template import create_template with self.assertRaises(TenantMissingError): create_template(tenant_id=None, data={"code": "no.tenant", "name": "无租户", "scope": "tenant"}, conn=self.conn) def test_12_template_unique_key_conflict(self): """同 (tenant_key, code, version) 重复创建 -> PblConflict。""" from pbl_blueprint.m1b.template import create_template payload = {"code": "tenant.dup", "name": "重复模板", "scope": "tenant", "version": "1.0.0"} create_template(tenant_id=TENANT, data=dict(payload), conn=self.conn) with self.assertRaises(PblConflict): create_template(tenant_id=TENANT, data=dict(payload), conn=self.conn) def test_13_cross_tenant_template_isolation(self): """A 租户模板对 B 租户不可见（隔离）。""" from pbl_blueprint.m1b.template import create_template, get_template row = create_template(tenant_id=TENANT, data={"code": "tenant.private", "name": "私有", "scope": "tenant", "version": "1.0.0"}, conn=self.conn) with self.assertRaises(PblNotFound): get_template(tenant_id=OTHER_TENANT, template_id=row["id"], conn=self.conn, allow_offline=False) # ---------- 4. 实例化 + 离线兜底 ---------- def test_14_instantiate_creates_blueprint_and_subobjects(self): """模板实例化真实产出蓝图 + 7 类子对象 + 扩展值（QC #4/#6）。""" from pbl_blueprint.m1b.template import instantiate_template plat = sql_rows( "SELECT id, code FROM pbl_blueprint_template WHERE tenant_id IS NULL " "ORDER BY code LIMIT 1", conn=self.conn)[0] res = instantiate_template( TENANT, template_id=plat["id"], actor_id="tester", conn=self.conn) self.assertTrue(res["ok"]) self.assertFalse(res["fallback"], "库内有模板，不应走离线兜底") self.assertTrue(res["blueprint"]["id"]) self.assertEqual(res["template"]["scope"], "platform") self.assertGreaterEqual(res["counts"]["__total__"], 1, "未实例化出任何子对象") self.assertGreaterEqual(res["counts"]["ext_written"], 1, "未写入任何扩展字段值") # 子对象行真实落库 for st in SUBOBJECT_TYPES: if res["counts"][st]: n = sql_scalar( "SELECT COUNT(*) FROM pbl_subobject_ext WHERE tenant_id = ? " "AND blueprint_id = ? AND subobject_type = ?", [TENANT, res["blueprint"]["id"], st], conn=self.conn) self.assertIsNotNone(n) def test_15_instantiate_requires_tenant(self): """实例化缺租户 -> TenantMissingError。""" from pbl_blueprint.m1b.template import instantiate_template with self.assertRaises(TenantMissingError): instantiate_template(None, code="platform.pbl.stem", conn=self.conn) def test_16_offline_fallback_marks_reason(self): """库中不存在的模板走离线兜底，且如实标注 source/fallback_reason（QC 要求）。""" from pbl_blueprint.m1b.template import get_template, load_offline_templates items = load_offline_templates(reason="unit-test forced") if not items: self.skipTest("离线模板包不存在，跳过兜底断言") self.assertEqual(items[0]["source"], "offline_fallback") self.assertIn("unit-test forced", items[0]["fallback_reason"]) self.assertFalse(items[0]["writable"], "离线兜底模板必须只读") hit = get_template(tenant_id=TENANT, code=items[0]["code"], version=items[0]["version"], conn=self.conn) self.assertEqual(hit["source"], "offline_fallback") def test_17_instantiate_from_offline_template(self): """离线模板也能完成实例化（DB 模板缺失时的兜底可用性）。""" from pbl_blueprint.m1b.template import ( instantiate_template, load_offline_templates, ) items = load_offline_templates(reason="unit-test forced") if not items: self.skipTest("离线模板包不存在") code = items[-1]["code"] # 确保库里没有该 code 的租户可见模板 res = instantiate_template(TENANT, code=code, version=items[-1]["version"], actor_id="tester", conn=self.conn) self.assertTrue(res["ok"]) self.assertTrue(res["fallback"]) self.assertTrue(res["fallback_reason"]) # ---------- 5. 租户上下文 ---------- def test_18_require_tenant_fail_closed(self): """require_tenant 对空/空白/'null' 一律抛错（fail-closed）。""" for bad in (None, "", " ", "null", "None"): with self.assertRaises(TenantMissingError): require_tenant(bad) self.assertEqual(require_tenant(TENANT), TENANT) self.assertIsNone(require_tenant(None, allow_null=True)) def test_19_m1b_status_reports_counts(self): """m1b_status 输出可核验的落库状态快照。""" st = m1b_init.m1b_status(conn=self.conn) self.assertTrue(all(st["tables"].values()), "存在未建出的表") self.assertGreater(st["counts"]["pbl_ext_field_def"], 0) self.assertGreater(st["counts"]["pbl_blueprint_template"], 0) self.assertEqual(st["subobject_types"], list(SUBOBJECT_TYPES)) # ---------- 6. API 层（pbl_agent_runtime 依赖面） ---------- def test_20_api_surface_available(self): """pbl_blueprint.api 暴露 pbl_template_instantiate / pbl_blueprint_create。""" from pbl_blueprint.m1b.api import ( M1B_API_REGISTRY, pbl_blueprint_create, pbl_template_instantiate, ) self.assertTrue(callable(pbl_template_instantiate)) self.assertTrue(callable(pbl_blueprint_create)) self.assertIn("pbl_template_instantiate", M1B_API_REGISTRY) self.assertIn("pbl_blueprint_create", M1B_API_REGISTRY) self.assertGreaterEqual(len(M1B_API_REGISTRY), 20) def test_21_api_returns_unified_envelope(self): """API 返回统一响应包，异常不外泄（ok/code/message/data）。""" from pbl_blueprint.m1b.api import pbl_template_instantiate res = pbl_template_instantiate(tenant_id=None) self.assertFalse(res["ok"]) self.assertEqual(res["code"], ErrorCode.TENANT_MISSING) self.assertIn("http_status", res) res2 = pbl_template_instantiate(tenant_id=TENANT, code="platform.pbl.stem") self.assertTrue(res2["ok"], res2) self.assertIn("counts", res2["data"]) def test_22_api_blueprint_create_direct_and_from_template(self): """pbl_blueprint_create 两种形态：直接建 / 由模板派生。""" from pbl_blueprint.m1b.api import pbl_blueprint_create r1 = pbl_blueprint_create(tenant_id=TENANT, data={"name": "直接创建的蓝图"}) self.assertTrue(r1["ok"], r1) self.assertTrue(r1["data"]["id"]) r2 = pbl_blueprint_create(tenant_id=TENANT, data={"template_code": "platform.pbl.stem", "name": "由模板派生"}) self.assertTrue(r2["ok"], r2) self.assertIn("subobjects", r2["data"]) if __name__ == "__main__": unittest.main(verbosity=2)