chongming-core/test/test_worker_isolation.py

"""
Worker 进程隔离性测试

验证进程插件与网关完全隔离：阻断 chongming（网关包）的两种导入路径后，
worker 侧全链路（SDK 导入 → 插件配置自读 → 插件 import → 契约校验）仍可用。

与容器行为一致：chdir 到插件项目根 workers/demo/（= 打包 zip 根），
worker_serve 即以该目录为 CWD 读配置、并把 CWD 加入 sys.path。

任何对 chongming.* / src.chongming.* 的导入都会直接触发 ImportError。
"""
import os
import sys
from pathlib import Path

SRC = Path(__file__).parent
REPO = SRC.parent

# 模拟容器：CWD = 插件项目根（zip 根），并把它置于 sys.path 首位
PROJECT = REPO / "workers" / "demo"
os.chdir(PROJECT)
sys.path.insert(0, str(PROJECT))
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
import chongming_worker.storage as sto_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")
subjects = [r["subject"] for r in routes]
check(
    "进程内契约校验通过（TOML ↔ 注册表 ↔ 模型）",
    len(routes) == 5 and "demo.hello" in subjects,
    f"routes={subjects}",
)

# ========== 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, sto_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 进程隔离性全部通过")