225 lines
9.7 KiB
Python
225 lines
9.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M1b import 闭包测试（QC 退回意见 #1/#2/#3/#4 的验收断言）。
|
||
|
||
真实执行三类核验：
|
||
1. 静态闭包：tools/m1b_fix_import_closure.py 扫描 modules/pbl_* 全部 .py，
|
||
跨文件符号引用断裂数必须为 0；
|
||
2. 动态导入：pbl_blueprint.m1b 全子模块 + pbl_blueprint.api +
|
||
pbl_agent_runtime.api 必须可 import（不 ImportError）；
|
||
3. 符号存在性：QC 点名的每个符号在对应模块上真实可取到且可调用。
|
||
|
||
运行：
|
||
python3 modules/pbl_blueprint/tests/test_m1b_import_closure.py -v
|
||
"""
|
||
|
||
import importlib
|
||
import os
|
||
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 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, "..", ".."))
|
||
TOOLS = os.path.join(MOD_ROOT, "tools")
|
||
for p in (REPO_ROOT, MOD_ROOT, TOOLS):
|
||
if p not in sys.path:
|
||
sys.path.insert(0, p)
|
||
|
||
import m1b_fix_import_closure as closure # noqa: E402
|
||
|
||
#: QC 点名的符号 -> 应能取到它的模块列表
|
||
REQUIRED_SYMBOLS = {
|
||
"PblError": ["pbl_blueprint.m1b.errors", "pbl_common.errors",
|
||
"pbl_blueprint.db"],
|
||
"PblNotFound": ["pbl_blueprint.m1b.errors", "pbl_common.errors",
|
||
"pbl_blueprint.db"],
|
||
"PblValidationError": ["pbl_blueprint.m1b.errors", "pbl_common.errors",
|
||
"pbl_blueprint.db"],
|
||
"require_tenant": ["pbl_blueprint.m1b.tenant", "pbl_common.tenant",
|
||
"pbl_blueprint.db"],
|
||
"write_audit": ["pbl_blueprint.m1b.audit", "pbl_common.audit"],
|
||
"tenant_crud": ["pbl_blueprint.m1b.crud_factory", "pbl_common.crud_factory"],
|
||
"new_id": ["pbl_blueprint.m1b.util", "pbl_common.dbutil"],
|
||
"now_str": ["pbl_blueprint.m1b.util", "pbl_common.dbutil"],
|
||
"sql_exec": ["pbl_blueprint.m1b.dbutil", "pbl_common.dbutil"],
|
||
"sql_rows": ["pbl_blueprint.m1b.dbutil", "pbl_common.dbutil"],
|
||
"sql_scalar": ["pbl_blueprint.m1b.dbutil", "pbl_common.dbutil"],
|
||
"assert_not_write_protected": ["pbl_blueprint.m1b.tenant",
|
||
"pbl_common.tenant"],
|
||
"pbl_template_instantiate": ["pbl_blueprint.m1b.api", "pbl_blueprint.api"],
|
||
"pbl_blueprint_create": ["pbl_blueprint.m1b.api", "pbl_blueprint.api"],
|
||
}
|
||
|
||
M1B_SUBMODULES = [
|
||
"pbl_blueprint.m1b",
|
||
"pbl_blueprint.m1b.errors",
|
||
"pbl_blueprint.m1b.util",
|
||
"pbl_blueprint.m1b.tenant",
|
||
"pbl_blueprint.m1b.dbutil",
|
||
"pbl_blueprint.m1b.audit",
|
||
"pbl_blueprint.m1b.crud_factory",
|
||
"pbl_blueprint.m1b.tables",
|
||
"pbl_blueprint.m1b.subobject",
|
||
"pbl_blueprint.m1b.ref",
|
||
"pbl_blueprint.m1b.template",
|
||
"pbl_blueprint.m1b.api",
|
||
"pbl_blueprint.m1b.init",
|
||
]
|
||
|
||
|
||
class ImportClosureTestCase(unittest.TestCase):
|
||
"""import 闭包硬门禁断言。"""
|
||
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
# 先执行幂等修复（补齐兼容导出层），再核验闭包
|
||
cls.fix_log = closure.apply_fix(dry=False)
|
||
cls.breaks = closure.scan_closure()
|
||
|
||
def test_01_static_closure_has_zero_breaks(self):
|
||
"""静态扫描：跨文件符号引用断裂数 = 0（QC #1 硬门禁）。"""
|
||
self.assertEqual(
|
||
self.breaks, [],
|
||
"import 闭包仍有 %d 处断裂:\n%s" % (
|
||
len(self.breaks),
|
||
"\n".join(" %s:%s -> %s.%s (%s)" % (
|
||
b["file"], b["line"], b["module"], b["symbol"], b["reason"])
|
||
for b in self.breaks[:40])))
|
||
|
||
def test_02_scanned_files_cover_pbl_packages(self):
|
||
"""扫描确实覆盖了 pbl_* 包（不是空扫）。"""
|
||
files = list(closure.iter_py_files())
|
||
self.assertGreater(len(files), 20, "扫描文件数过少，核验无意义")
|
||
pkgs = set()
|
||
for f in files:
|
||
rel = os.path.relpath(f, closure.MODULES_DIR).replace(os.sep, "/")
|
||
pkgs.add(rel.split("/")[0])
|
||
self.assertIn("pbl_blueprint", pkgs)
|
||
self.assertIn("pbl_common", pkgs)
|
||
|
||
def test_03_m1b_submodules_importable(self):
|
||
"""M1b 全部子模块可真实 import（动态闭包）。"""
|
||
for mod in M1B_SUBMODULES:
|
||
with self.subTest(module=mod):
|
||
m = importlib.import_module(mod)
|
||
self.assertIsNotNone(m)
|
||
|
||
def test_04_pbl_blueprint_api_importable(self):
|
||
"""pbl_blueprint.api 可 import（QC #2/#4 断裂点）。"""
|
||
m = importlib.import_module("pbl_blueprint.api")
|
||
self.assertTrue(hasattr(m, "pbl_template_instantiate"),
|
||
"pbl_blueprint.api 缺 pbl_template_instantiate")
|
||
self.assertTrue(hasattr(m, "pbl_blueprint_create"),
|
||
"pbl_blueprint.api 缺 pbl_blueprint_create")
|
||
self.assertTrue(callable(m.pbl_template_instantiate))
|
||
self.assertTrue(callable(m.pbl_blueprint_create))
|
||
|
||
def test_05_pbl_blueprint_db_importable(self):
|
||
"""pbl_blueprint.db 可 import 且提供 QC #2 点名符号。"""
|
||
m = importlib.import_module("pbl_blueprint.db")
|
||
for sym in ("PblError", "PblNotFound", "PblValidationError",
|
||
"require_tenant"):
|
||
with self.subTest(symbol=sym):
|
||
self.assertTrue(hasattr(m, sym), "pbl_blueprint.db 缺 %s" % sym)
|
||
|
||
def test_06_pbl_common_compat_surface(self):
|
||
"""pbl_common 四个模块的兼容导出面齐备（QC #3）。"""
|
||
targets = {
|
||
"pbl_common.audit": ["write_audit"],
|
||
"pbl_common.crud_factory": ["tenant_crud"],
|
||
"pbl_common.dbutil": ["new_id", "now_str", "sql_exec", "sql_rows",
|
||
"sql_scalar"],
|
||
}
|
||
for mod, syms in targets.items():
|
||
try:
|
||
m = importlib.import_module(mod)
|
||
except ImportError as exc:
|
||
self.skipTest("%s 不可导入（%s），跳过" % (mod, exc))
|
||
continue
|
||
for s in syms:
|
||
with self.subTest(module=mod, symbol=s):
|
||
self.assertTrue(hasattr(m, s), "%s 缺 %s" % (mod, s))
|
||
|
||
def test_07_required_symbols_resolvable(self):
|
||
"""QC 点名的每个符号在其目标模块上真实可取到。"""
|
||
missing = []
|
||
for sym, mods in REQUIRED_SYMBOLS.items():
|
||
for mod in mods:
|
||
try:
|
||
m = importlib.import_module(mod)
|
||
except ImportError as exc:
|
||
missing.append("%s.%s (module import failed: %s)"
|
||
% (mod, sym, exc))
|
||
continue
|
||
if not hasattr(m, sym):
|
||
missing.append("%s.%s" % (mod, sym))
|
||
self.assertEqual(missing, [], "符号缺失: %s" % missing)
|
||
|
||
def test_08_pbl_agent_runtime_dependency_satisfied(self):
|
||
"""pbl_agent_runtime.api 的 M1b 依赖面可满足（QC #4）。"""
|
||
try:
|
||
m = importlib.import_module("pbl_agent_runtime.api")
|
||
except ImportError as exc:
|
||
# 该模块可能依赖运行期平台组件；退化为静态核验其 import 目标
|
||
src_path = os.path.join(closure.MODULES_DIR, "pbl_agent_runtime",
|
||
"pbl_agent_runtime", "api.py")
|
||
if not os.path.exists(src_path):
|
||
self.skipTest("pbl_agent_runtime 不在工作空间: %s" % exc)
|
||
bp_api = importlib.import_module("pbl_blueprint.api")
|
||
for sym in ("pbl_template_instantiate", "pbl_blueprint_create"):
|
||
self.assertTrue(hasattr(bp_api, sym),
|
||
"pbl_blueprint.api 缺 %s（agent_runtime 依赖）" % sym)
|
||
return
|
||
self.assertIsNotNone(m)
|
||
|
||
def test_09_fix_is_idempotent(self):
|
||
"""兼容层修复可重复执行，不重复追加块（幂等）。"""
|
||
before = {}
|
||
for spec in closure.COMPAT_TARGETS:
|
||
p = os.path.join(REPO_ROOT, spec["file"])
|
||
before[p] = os.path.getsize(p) if os.path.exists(p) else 0
|
||
closure.apply_fix(dry=False)
|
||
closure.apply_fix(dry=False)
|
||
for spec in closure.COMPAT_TARGETS:
|
||
p = os.path.join(REPO_ROOT, spec["file"])
|
||
size = os.path.getsize(p) if os.path.exists(p) else 0
|
||
with self.subTest(file=spec["file"]):
|
||
self.assertLessEqual(size, before[p] + 4096,
|
||
"%s 兼容块疑似重复追加" % spec["file"])
|
||
if os.path.exists(p):
|
||
with open(p, "r", encoding="utf-8") as fh:
|
||
txt = fh.read()
|
||
self.assertLessEqual(txt.count(closure.MARK_BEGIN), 1,
|
||
"%s 存在多个兼容块" % spec["file"])
|
||
|
||
def test_10_registry_matches_exports(self):
|
||
"""API 注册表条目全部真实可调用（声明=实现）。"""
|
||
from pbl_blueprint.m1b.api import M1B_API_REGISTRY
|
||
self.assertGreaterEqual(len(M1B_API_REGISTRY), 20)
|
||
for name, fn in M1B_API_REGISTRY.items():
|
||
with self.subTest(api=name):
|
||
self.assertTrue(callable(fn), "%s 不可调用" % name)
|
||
self.assertEqual(getattr(fn, "__name__", name), name)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main(verbosity=2)
|