126 lines
6.9 KiB
Python
126 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""import_closure 门禁离线单测（复刻 pbls pbl_common 半迁移事故形状）。"""
|
||
import os
|
||
import shutil
|
||
import sys
|
||
|
||
# 本地无 ahserver 等平台依赖 → 绕开包 __init__，按文件直接加载被测模块
|
||
import importlib.util as _ilu
|
||
_spec = _ilu.spec_from_file_location(
|
||
"import_closure",
|
||
os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"pipeline_service", "import_closure.py"))
|
||
_ic = _ilu.module_from_spec(_spec)
|
||
_spec.loader.exec_module(_ic)
|
||
check_closure_for_delivery = _ic.check_closure_for_delivery
|
||
closure_report = _ic.closure_report
|
||
|
||
ROOT = "/tmp/ic_test"
|
||
FAILS = []
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
print(("PASS" if cond else "FAIL"), name, detail[:160])
|
||
if not cond:
|
||
FAILS.append(name)
|
||
|
||
|
||
def w(path, content):
|
||
full = os.path.join(ROOT, path)
|
||
os.makedirs(os.path.dirname(full), exist_ok=True)
|
||
with open(full, "w", encoding="utf-8") as f:
|
||
f.write(content)
|
||
|
||
|
||
# ── 用例 1：事故形状复刻——重写 errors 删旧符号，内部+依赖仍 import 旧名 ──
|
||
shutil.rmtree(ROOT, ignore_errors=True)
|
||
w("modules/pbl_common/pbl_common/__init__.py", "from .kernel import PblError\n")
|
||
w("modules/pbl_common/pbl_common/errors.py",
|
||
"class PBLError(Exception):\n pass\n\n\ndef err(cat, key):\n return {}\n\n\ndef fail(cat, key):\n return {}\n")
|
||
w("modules/pbl_common/pbl_common/kernel.py", "class PblError(Exception):\n code = 'X'\n")
|
||
w("modules/pbl_common/pbl_common/context.py",
|
||
"from pbl_common.errors import TenantMissingError, ErrorCode\n")
|
||
w("modules/pbl_validation/pbl_validation/__init__.py", "")
|
||
w("modules/pbl_validation/pbl_validation/api.py",
|
||
"from pbl_common.errors import fail\nfrom pbl_common.audit import write_audit\n")
|
||
|
||
# 本任务写了 pbl_common 的 errors.py（重写）→ 门禁应抓到 context.py 断裂
|
||
written = [os.path.join(ROOT, "modules/pbl_common/pbl_common/errors.py")]
|
||
probs, involved = check_closure_for_delivery(ROOT, written)
|
||
check("事故形状被拦截", probs is not None and len(probs) >= 2,
|
||
"problems=%s involved=%s" % (probs and len(probs), involved))
|
||
check("涉及包=pbl_common", involved == {"pbl_common"}, str(involved))
|
||
syms = {(p["file"], p["symbol"]) for p in (probs or [])}
|
||
check("抓到 context.py TenantMissingError",
|
||
("modules/pbl_common/pbl_common/context.py", "TenantMissingError") in syms, str(syms))
|
||
check("抓到依赖方 pbl_validation missing-module",
|
||
any(p["kind"] == "missing-module" and p["symbol"] == "pbl_common.audit"
|
||
for p in (probs or [])), str(probs))
|
||
rep = closure_report(ROOT, written)
|
||
check("closure_report 非空含断裂", "断裂" in rep and "ImportError" in rep, rep[:80])
|
||
|
||
# ── 用例 2：未触碰的包存量断裂不拦（防死锁）──
|
||
w("modules/pbl_unrelated/pbl_unrelated/__init__.py", "")
|
||
w("modules/pbl_unrelated/pbl_unrelated/api.py", "from pbl_common.nothere import X\n")
|
||
probs2, involved2 = check_closure_for_delivery(ROOT, written)
|
||
check("未触碰包不在 involved", "pbl_unrelated" not in involved2, str(involved2))
|
||
# 但它 import pbl_common（涉及包）→ 应作为引用方被核验
|
||
check("引用涉及包的文件被核验",
|
||
any(p["file"].startswith("modules/pbl_unrelated") for p in (probs2 or [])),
|
||
str(probs2))
|
||
|
||
# ── 用例 3：修复后（补兼容符号）门禁放行 ──
|
||
w("modules/pbl_common/pbl_common/errors.py",
|
||
"class PBLError(Exception):\n pass\n\n\ndef err(cat, key):\n return {}\n\n\ndef fail(cat, key):\n return {}\n\n\n"
|
||
"# 兼容层（设计文档§5 向后兼容）\n"
|
||
"class PblError(PBLError):\n code = 'PBL_E'\n\n\nclass TenantMissingError(PblError):\n pass\n\n\nclass TenantInvalidError(PblError):\n pass\n\n\nclass ErrorCode(object):\n TENANT_MISSING = 'PBL-TENANT-0001'\n")
|
||
w("modules/pbl_common/pbl_common/audit.py", "def write_audit(*a, **kw):\n return ''\n")
|
||
# context.py 引用修复 + unrelated 的 nothere 仍在 → 只应剩 unrelated 断裂
|
||
probs3, _ = check_closure_for_delivery(ROOT, written)
|
||
remain = [p for p in (probs3 or []) if not p["file"].startswith("modules/pbl_unrelated")]
|
||
check("补兼容符号后 pbl_common 闭包干净", not remain, str(remain))
|
||
|
||
# ── 用例 4：逃逸阀——__getattr__ 动态模块不误伤 ──
|
||
w("modules/pbl_dyn/pbl_dyn/__init__.py", "def __getattr__(name):\n return None\n")
|
||
w("modules/pbl_dyn/pbl_dyn/sub.py", "X = 1\n")
|
||
w("modules/pbl_validation/pbl_validation/dyn_user.py", "from pbl_dyn import WHATEVER\n")
|
||
probs4, _ = check_closure_for_delivery(
|
||
ROOT, [os.path.join(ROOT, "modules/pbl_dyn/pbl_dyn/sub.py")])
|
||
check("动态模块逃逸阀", not any("dyn" in p["file"] for p in (probs4 or [])), str(probs4))
|
||
|
||
# ── 用例 5：语法坏文件跳过（归空壳门禁）不崩 ──
|
||
w("modules/pbl_validation/pbl_validation/broken.py", "def broken(:\n")
|
||
probs5, _ = check_closure_for_delivery(
|
||
ROOT, [os.path.join(ROOT, "modules/pbl_validation/pbl_validation/broken.py")])
|
||
check("语法坏文件不崩", probs5 is not None, str(probs5))
|
||
|
||
# ── 用例 6：无 .py 写入（纯文档任务）零成本放行 ──
|
||
probs6, involved6 = check_closure_for_delivery(ROOT, [os.path.join(ROOT, "docs/x.md")])
|
||
check("纯文档任务不拦", probs6 == [] and not involved6, str((probs6, involved6)))
|
||
|
||
# ── 用例 7：相对导入 + 别名 + 星号 ──
|
||
w("modules/pbl_rel/pbl_rel/__init__.py", "from .core import helper\n__all__ = ['helper']\n")
|
||
w("modules/pbl_rel/pbl_rel/core.py", "def helper():\n return 1\n")
|
||
w("modules/pbl_rel/pbl_rel/use.py", "from . import helper\nfrom .core import helper as h2\nfrom pbl_rel.core import *\n")
|
||
probs7, _ = check_closure_for_delivery(ROOT, [os.path.join(ROOT, "modules/pbl_rel/pbl_rel/use.py")])
|
||
check("相对导入/别名/星号解析", not any("pbl_rel" in p["file"] for p in (probs7 or [])), str(probs7))
|
||
w("modules/pbl_rel/pbl_rel/bad.py", "from . import NOT_THERE\n")
|
||
probs8, _ = check_closure_for_delivery(ROOT, [os.path.join(ROOT, "modules/pbl_rel/pbl_rel/bad.py")])
|
||
check("相对导入断裂也抓", any(p["symbol"] == "NOT_THERE" for p in (probs8 or [])), str(probs8))
|
||
|
||
# ── 用例 8：仓库归属——无包目录的仓库（modules/pbls 形态）入口断裂也拦 ──
|
||
w("modules/pbls/app/pbls.py", "from world.api import load_world\nfrom pbl_common.kernel import PblError\n")
|
||
w("modules/world/world/__init__.py", "def load_world(env=None):\n return True\n")
|
||
w("modules/world/__init__.py", "from .world import load_world\n")
|
||
probs9, involved9 = check_closure_for_delivery(
|
||
ROOT, [os.path.join(ROOT, "modules/pbls/app/pbls.py")])
|
||
check("无包仓库入口纳入核验", "pbls" in involved9, str(involved9))
|
||
check("抓到 world.api 断裂", any(p["symbol"] == "world.api" for p in (probs9 or [])),
|
||
str(probs9))
|
||
check("pbl_common.kernel 存在不误报", not any("pbl_common.kernel" in str(p) for p in (probs9 or [])),
|
||
str(probs9))
|
||
|
||
print("\nFAILS:", FAILS or "NONE")
|
||
sys.exit(1 if FAILS else 0)
|