#!/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)