"""round-11 集成补洞：执行器接 nspawn 单元（P2 差距表 §4）。 只测 _guest_host_command 的命令构造与 CHV_CELL_ENABLED 开关： 单元开（默认）时 host.py 调用改为 sudo chv-cell run -- python3 /app/host.py --state … --codex-home … --sandbox danger-full-access …（sudo 不可省：chv-cell 须 root，执行器以 chv 身份调；round-11 真机修复）；单元关时走原命令；非法开关值即 ValueError。只用 mock，不启动真实 VM、不连真实网络。 """ import os import sys import tempfile import unittest from pathlib import Path try: from cloud.chv import CloudHypervisorExecutor except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.chv import CloudHypervisorExecutor class FakeResult: def __init__(self, returncode=0, stdout="", stderr=""): self.returncode = returncode self.stdout = stdout self.stderr = stderr class FakeProc: def __init__(self, alive=True): self._alive = alive def poll(self): return None if self._alive else 1 def _write(path, data=b"x"): with open(path, "wb") as fh: fh.write(data) return path class CellCase(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.dir = self.tmp.name self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) self.kernel = _write(os.path.join(self.dir, "vmlinux")) self.rootfs = _write(os.path.join(self.dir, "disk.raw"), b"R" * 100) self.ssh_key = _write(os.path.join(self.dir, "id_chv")) self.kvm = _write(os.path.join(self.dir, "kvm")) self.cpuinfo = os.path.join(self.dir, "cpuinfo") with open(self.cpuinfo, "w") as fh: fh.write("flags : fpu vmx sse\n") self.state_dir = os.path.join(self.dir, "state") def tearDown(self): self.tmp.cleanup() def make_exec(self, env_extra=None): env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", "CHV_HOST_KEY_MODE": "keyscan", "CHV_SSH_FORWARD": "1"} if env_extra: env.update(env_extra) return CloudHypervisorExecutor( binary=self.binary, ch_remote=self.ch_remote, kernel=self.kernel, rootfs=self.rootfs, ssh_key=self.ssh_key, kvm_path=self.kvm, cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, runner=lambda cmd, timeout=None: FakeResult(0, "", ""), launcher=lambda cmd: FakeProc(alive=True), api_client=lambda *a, **k: (200, {}), sleeper=lambda s: None, env=env) class CellCommandTest(CellCase): def test_cell_on_by_default(self): # 开关未设即默认开：完整命令与契约示例一致。 ex = self.make_exec() cmd = ex._guest_host_command("status", "--task", "t1") self.assertEqual(cmd, [ "sudo", "chv-cell", "run", "--", "python3", "/app/host.py", "--state", "/home/chv/.codex/host-state.json", "--codex-home", "/home/chv/.codex", "--sandbox", "danger-full-access", "status", "--task", "t1"]) def test_cell_on_explicit_1(self): ex = self.make_exec(env_extra={"CHV_CELL_ENABLED": "1"}) cmd = ex._guest_host_command("status", "--task", "t1") self.assertEqual(cmd[:4], ["sudo", "chv-cell", "run", "--"]) self.assertIn("--sandbox", cmd) self.assertEqual( cmd[cmd.index("--sandbox") + 1], "danger-full-access") def test_cell_off_returns_original_command(self): # 关时走原命令：一个 token 也不多不少（旧镜像逃生口）。 ex = self.make_exec(env_extra={"CHV_CELL_ENABLED": "0"}) cmd = ex._guest_host_command("status", "--task", "t1") self.assertEqual(cmd, ["python3", "/app/host.py", "status", "--task", "t1"]) self.assertNotIn("chv-cell", cmd) self.assertNotIn("--state", cmd) self.assertNotIn("--codex-home", cmd) self.assertNotIn("--sandbox", cmd) def test_cell_on_preserves_original_params(self): # 原参数一个不少：model/provider/effort/approval/ # 调用方 sandbox 与子命令全部保留且顺序不变；强制的 # danger-full-access 放全局参数最后，盖过调用方的 # read-only（argparse 后胜出），read-only 本身仍在。 ex = self.make_exec() cmd = ex._guest_host_command( "dispatch", "--name", "t", "--task", "x", model="m1", effort="low", approval="ask", model_provider="sub2api", sandbox="read-only") self.assertEqual(cmd, [ "sudo", "chv-cell", "run", "--", "python3", "/app/host.py", "--state", "/home/chv/.codex/host-state.json", "--codex-home", "/home/chv/.codex", "--model", "m1", "--model-provider", "sub2api", "--effort", "low", "--approval-policy", "untrusted", "--sandbox", "read-only", "--sandbox", "danger-full-access", "dispatch", "--name", "t", "--task", "x"]) # --state/--codex-home 紧跟 host.py 路径。 host_idx = cmd.index("/app/host.py") self.assertEqual( cmd[host_idx + 1:host_idx + 5], ["--state", "/home/chv/.codex/host-state.json", "--codex-home", "/home/chv/.codex"]) def test_sandbox_only_in_cell_path(self): # 调用方没给 sandbox、approval 也没带时：单元路径 # 有且仅有一个 --sandbox（danger-full-access）， # 非单元路径一个也没有。 cell_on = self.make_exec() on_cmd = cell_on._guest_host_command("status", "--task", "t1") self.assertEqual(on_cmd.count("--sandbox"), 1) self.assertEqual( on_cmd[on_cmd.index("--sandbox") + 1], "danger-full-access") cell_off = self.make_exec( env_extra={"CHV_CELL_ENABLED": "0"}) off_cmd = cell_off._guest_host_command( "status", "--task", "t1") self.assertEqual(off_cmd.count("--sandbox"), 0) def test_cell_invalid_value_raises(self): # 只认 0/1：非法值即 ValueError，不静默 fallback。 for bad in ("yes", "2", "", " true"): ex = self.make_exec( env_extra={"CHV_CELL_ENABLED": bad}) with self.assertRaises(ValueError): ex._guest_host_command("status", "--task", "t1") def test_cell_wiring_reaches_ssh_command(self): # 接线落到 SSH 层：build_ssh_command 的远端命令里 # sudo chv-cell 前缀在、python3 在其后。 ex = self.make_exec() inner = ex._guest_host_command("status", "--task", "t1") cmd = ex.build_ssh_command({"id": "t1"}, inner) self.assertIn("sudo", cmd) self.assertIn("chv-cell", cmd) self.assertIn("danger-full-access", cmd) self.assertLess(cmd.index("sudo"), cmd.index("chv-cell")) self.assertLess(cmd.index("chv-cell"), cmd.index("python3")) if __name__ == "__main__": unittest.main()