350 lines
17 KiB
Python
350 lines
17 KiB
Python
#!/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)
|