chongming-core/test/test_build_plugin.py

"""
build_plugin 打包命令冒烟测试

验证进程插件打包产物的完整性与错误路径：
  1. demo 插件打包成功，zip = 项目根直接压缩（代码/配置/最小根配置/依赖清单/可执行 start.sh）
  2. 打包排除 __pycache__ / .venv / *.pyc 等非产物内容
  3. 插件项目目录缺失时报 FileNotFoundError
  4. 插件项目内配置缺失时报 FileNotFoundError

运行: uv run python test_build_plugin.py
"""
import shutil
import sys
import tempfile
import tomllib
import zipfile
from pathlib import Path

SRC = Path(__file__).parent
sys.path.insert(0, str(SRC.parent / "utils" / "cli" / "src"))

from chongming_cli import build_plugin as bp  # noqa: E402
from chongming_cli import gen_plugin as gp  # noqa: E402

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)


tmp = Path(tempfile.mkdtemp(prefix="build_plugin_test_"))


# ========== 1. demo 插件打包 ==========
zip_path = bp.build_plugin_package("demo", output_dir=tmp / "dist")
check("zip 产物生成", zip_path.exists(), str(zip_path))

with zipfile.ZipFile(zip_path) as z:
    names = set(z.namelist())
    check("包含插件代码 workers/demo/__init__.py", "workers/demo/__init__.py" in names)
    check("包含插件配置 config/workers/demo/config.toml", "config/workers/demo/config.toml" in names)
    check("包含 Zenoh 配置", "config/zenoh_config.json5" in names)
    check("包含最小根配置 config.toml", "config.toml" in names)
    check("包含依赖清单 pyproject.toml", "pyproject.toml" in names)
    check("包含 start.sh", "start.sh" in names)
    check("不包含 __pycache__", not any("__pycache__" in n for n in names))

    # 最小根配置：worker 启动需 [zenoh] 节，且不含网关侧敏感配置
    root_conf = tomllib.loads(z.read("config.toml").decode())
    check("根配置含 [zenoh] 节", "zenoh" in root_conf, str(sorted(root_conf)))
    check(
        "根配置不泄露网关 [security] 密钥",
        "secret_key" not in z.read("config.toml").decode(),
    )

    # 依赖清单：chongming-worker 内网 git 源 + worker_serve 入口
    pyproject = tomllib.loads(z.read("pyproject.toml").decode())
    deps = pyproject["project"]["dependencies"]
    check("pyproject 依赖包含 chongming-worker", "chongming-worker" in deps)
    check("pyproject 依赖包含 orjson（SDK 运行时依赖）", "orjson" in deps)
    check(
        "pyproject 声明 worker_serve 入口",
        pyproject["project"]["scripts"].get("worker_serve") == "chongming_worker.process:worker_serve",
    )
    check(
        "pyproject 声明 chongming-worker git 源",
        pyproject["tool"]["uv"]["sources"]["chongming-worker"]["git"] == gp.CHONGMING_WORKER_GIT,
    )

    # start.sh：可执行 + 指向正确的插件 type
    start_sh = z.read("start.sh").decode()
    check("start.sh 以 --type demo 启动", "--type demo" in start_sh)
    perms = (z.getinfo("start.sh").external_attr >> 16) & 0o777
    check("start.sh 可执行（0o755）", perms == 0o755, oct(perms))


# ========== 2. 打包排除非产物内容（临时最小项目） ==========
def make_mini_project(proj: Path) -> None:
    """构造一个结构合法的迷你插件项目（含需要排除的垃圾文件）。"""
    (proj / "workers" / "mini" / "router").mkdir(parents=True)
    (proj / "workers" / "mini" / "__init__.py").write_text("", encoding="utf-8")
    (proj / "workers" / "mini" / "router" / "__init__.py").write_text("", encoding="utf-8")
    conf = proj / "config" / "workers" / "mini"
    conf.mkdir(parents=True)
    (conf / "config.toml").write_text(
        '[workers]\ntype = "mini"\nmodule = "workers.mini"\nload_mode = "external"\n',
        encoding="utf-8",
    )
    (proj / "config" / "zenoh_config.json5").write_text("{}", encoding="utf-8")
    (proj / "config.toml").write_text('[zenoh]\njson5_path = "config/zenoh_config.json5"\n', encoding="utf-8")
    (proj / "pyproject.toml").write_text("[project]\nname = \"mini-worker\"\n", encoding="utf-8")
    (proj / "start.sh").write_text("#!/usr/bin/env bash\necho mini\n", encoding="utf-8")
    (proj / "start.sh").chmod(0o755)
    # 垃圾内容：缓存目录、字节码、虚拟环境
    (proj / "workers" / "mini" / "__pycache__" / "x.pyc").parent.mkdir(parents=True)
    (proj / "workers" / "mini" / "__pycache__" / "x.pyc").write_bytes(b"x")
    (proj / "workers" / "mini" / "y.pyc").write_bytes(b"y")
    (proj / ".venv" / "bin" / "pip").parent.mkdir(parents=True)
    (proj / ".venv" / "bin" / "pip").write_text("", encoding="utf-8")


orig_workers = bp.WORKERS_DIR
mini_proj = tmp / "workers" / "mini"
make_mini_project(mini_proj)
bp.WORKERS_DIR = tmp / "workers"
try:
    zip_path = bp.build_plugin_package("mini", output_dir=tmp / "dist")
    with zipfile.ZipFile(zip_path) as z:
        names = set(z.namelist())
        check("排除 __pycache__", not any("__pycache__" in n for n in names), str(sorted(names)))
        check("排除 *.pyc", not any(n.endswith(".pyc") for n in names))
        check("排除 .venv", not any(n.startswith(".venv") for n in names))
        check("保留代码与配置", {"workers/mini/__init__.py", "config/workers/mini/config.toml"} <= names)
finally:
    bp.WORKERS_DIR = orig_workers


# ========== 3. 错误路径：插件项目目录缺失 ==========
bp.WORKERS_DIR = tmp / "no_workers"
try:
    bp.build_plugin_package("ghost", output_dir=tmp / "dist")
    check("缺插件项目目录应抛 FileNotFoundError", False)
except FileNotFoundError:
    check("缺插件项目目录应抛 FileNotFoundError", True)
finally:
    bp.WORKERS_DIR = orig_workers


# ========== 4. 错误路径：项目内插件配置缺失 ==========
# 只建代码目录、不建 config/workers/<name>/config.toml
(orig_workers / "ghost").mkdir(parents=True)
try:
    bp.build_plugin_package("ghost", output_dir=tmp / "dist")
    check("缺插件配置应抛 FileNotFoundError", False)
except FileNotFoundError:
    check("缺插件配置应抛 FileNotFoundError", True)
finally:
    shutil.rmtree(orig_workers / "ghost", ignore_errors=True)


# ========== 5. 错误路径：插件名不合法 ==========
try:
    bp.build_plugin_package("Bad-Name", output_dir=tmp / "dist")
    check("非法插件名应抛 ValueError", False)
except ValueError:
    check("非法插件名应抛 ValueError", True)


shutil.rmtree(tmp, ignore_errors=True)

print("\n" + ("=" * 40))
if failures:
    print(f"❌ {len(failures)} 项未通过: {failures}")
    sys.exit(1)
print("✅ 全部通过")