92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
|
"""
|
|||
|
|
Worker 进程隔离性测试
|
|||
|
|
|
|||
|
|
验证进程插件与网关完全隔离：阻断 chongming（网关包）的两种导入路径后，
|
|||
|
|
worker 侧全链路（SDK 导入 → 插件配置自读 → 插件 import → 契约校验）仍可用。
|
|||
|
|
|
|||
|
|
任何对 chongming.* / src.chongming.* 的导入都会直接触发 ImportError。
|
|||
|
|
"""
|
|||
|
|
import sys
|
|||
|
|
from pathlib import Path
|
|||
|
|
|
|||
|
|
SRC = Path(__file__).parent
|
|||
|
|
sys.path.insert(0, str(SRC))
|
|||
|
|
|
|||
|
|
# 阻断网关包导入（两种访问路径）。chongming_worker / zenoh / pydantic 等不受影响。
|
|||
|
|
sys.modules["chongming"] = None # type: ignore[assignment]
|
|||
|
|
sys.modules["src.chongming"] = None # type: ignore[assignment]
|
|||
|
|
|
|||
|
|
failures = []
|
|||
|
|
|
|||
|
|
|
|||
|
|
def check(name, cond, detail=""):
|
|||
|
|
status = "PASS" if cond else "FAIL"
|
|||
|
|
print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
|
|||
|
|
if not cond:
|
|||
|
|
failures.append(name)
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ========== 1. worker SDK 导入（若依赖网关包将在此抛 ImportError） ==========
|
|||
|
|
import chongming_worker.contracts as con_mod # noqa: E402
|
|||
|
|
import chongming_worker.process as proc_mod # noqa: E402
|
|||
|
|
import chongming_worker.route_config as rc_mod # noqa: E402
|
|||
|
|
import chongming_worker.routes_publisher as pub_mod # noqa: E402
|
|||
|
|
|
|||
|
|
check("worker SDK 导入（阻断 chongming 后成功）", True)
|
|||
|
|
|
|||
|
|
# ========== 2. 插件配置自读（不依赖网关 _discover_plugins） ==========
|
|||
|
|
wc = proc_mod.load_plugin_config("demo")
|
|||
|
|
check(
|
|||
|
|
"插件配置自读（type/load_mode/module）",
|
|||
|
|
wc["type"] == "demo"
|
|||
|
|
and wc["load_mode"] == "external"
|
|||
|
|
and wc["module"] == "workers.demo",
|
|||
|
|
f"type={wc['type']}, load_mode={wc['load_mode']}, module={wc['module']}",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ========== 3. 完整 worker 启动前置链路：import 插件 + 契约校验 ==========
|
|||
|
|
import workers.demo # noqa: E402,F401 触发 @rpc_method 注册
|
|||
|
|
|
|||
|
|
routes = con_mod.validate_worker_contracts("demo")
|
|||
|
|
check(
|
|||
|
|
"进程内契约校验通过（TOML ↔ 注册表 ↔ 模型）",
|
|||
|
|
len(routes) == 1 and routes[0]["subject"] == "demo.hello",
|
|||
|
|
f"routes={[r['subject'] for r in routes]}",
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
# ========== 4. 代码级双重保险：SDK 无对网关包的 import 语句（AST 精确检查） ==========
|
|||
|
|
import ast # noqa: E402
|
|||
|
|
import inspect # noqa: E402
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _imports_gateway(mod) -> list:
|
|||
|
|
"""返回模块中引用 chongming（非 chongming_worker）的 import 节点列表"""
|
|||
|
|
found = []
|
|||
|
|
tree = ast.parse(inspect.getsource(mod))
|
|||
|
|
for node in ast.walk(tree):
|
|||
|
|
if isinstance(node, ast.Import):
|
|||
|
|
for alias in node.names:
|
|||
|
|
if alias.name == "chongming" or alias.name.startswith("chongming."):
|
|||
|
|
found.append(alias.name)
|
|||
|
|
elif isinstance(node, ast.ImportFrom):
|
|||
|
|
if node.module and (
|
|||
|
|
node.module == "chongming" or node.module.startswith("chongming.")
|
|||
|
|
):
|
|||
|
|
found.append(node.module)
|
|||
|
|
return found
|
|||
|
|
|
|||
|
|
|
|||
|
|
for mod in (proc_mod, con_mod, pub_mod, rc_mod):
|
|||
|
|
bad = _imports_gateway(mod)
|
|||
|
|
if bad:
|
|||
|
|
check(f"{mod.__name__} import 了网关包: {bad}", False)
|
|||
|
|
break
|
|||
|
|
else:
|
|||
|
|
check("worker SDK 无对网关包的 import 语句", True)
|
|||
|
|
|
|||
|
|
print("\n" + ("=" * 40))
|
|||
|
|
if failures:
|
|||
|
|
print(f"❌ {len(failures)} 项未通过: {failures}")
|
|||
|
|
sys.exit(1)
|
|||
|
|
print("✅ Worker 进程隔离性全部通过")
|