"""CHV 收尾轨（feat-chv-closeup）新增测试。 只补上一轨 coverage 下 cloud/chv/executor.py 仍未覆盖的 深层生命周期分支：唤醒失败（vm.create/vm.boot/API 超时/ guest 就绪超时/host key 不匹配/provisioned 缺钉死）、 恢复与快照失败（restore 失败关 VM、快照未落盘/配额/ 总量剪枝）、取消与清理（cancel 真关、recycle、残留 socket/登记/磁盘/密钥扫描、脏标记）、中途死亡分类 （异常与 failed 状态改判 interrupted）、只读 execute 归类、入口校验、preflight 事后改写校验、worker_adapter 看门狗/调和的剩余分支。 全部用标准库与 mock runner/launcher/api_client，绝不启动 真实 VM（本机无 /dev/kvm，真机验证仍是 §13.2 的 V-1～V-9， 不在本文件声称已验证）。 """ import inspect import json import os import sys import tempfile import threading import unittest from pathlib import Path from unittest import mock try: from cloud.chv.executor import CloudHypervisorExecutor from cloud.chv.hostkeys import dirty_marker_path from cloud.chv.worker_adapter import ( call_goal_resume_breakpoint, cancel_watchdog_loop, quota_watchdog_loop, reconcile_warm_pool, watchdog_vm_name, ) from cloud.sandbox import ChvFailure except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.chv.executor import CloudHypervisorExecutor from cloud.chv.hostkeys import dirty_marker_path from cloud.chv.worker_adapter import ( call_goal_resume_breakpoint, cancel_watchdog_loop, quota_watchdog_loop, reconcile_warm_pool, watchdog_vm_name, ) from cloud.sandbox import ChvFailure 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 self.terminated = False self.killed = False def poll(self): return None if self._alive else 1 def terminate(self): self.terminated = True self._alive = False def wait(self, timeout=None): return 0 def kill(self): self.killed = True self._alive = False def _write(path, content=b"x"): with open(path, "wb") as fh: fh.write(content) os.chmod(path, 0o755) return path class CloseupCase(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") self.calls = [] self.api_calls = [] self.launches = [] self.audit = [] self.proc = None def tearDown(self): self.tmp.cleanup() def make_exec(self, runner=None, api_client=None, launcher=None, env_extra=None, **overrides): 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) kwargs = dict( 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=runner or self._default_runner, launcher=launcher or self._default_launcher, api_client=api_client or self._default_api, sleeper=lambda s: None, env=env) kwargs.update(overrides) ex = CloudHypervisorExecutor(**kwargs) ex.audit_callback = lambda e, t, d, k=None: self.audit.append( (e, t.get("id") if isinstance(t, dict) else t, d, k)) return ex def _default_runner(self, cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh": if cmd[-1] == "true": return FakeResult(0, "", "") if "dispatch" in cmd: return FakeResult(0, json.dumps( {"status": "completed", "text": "ok-dispatch"}), "") if "goal" in cmd or "goal-resume" in cmd: return FakeResult(0, "goal met after rounds\n", "") if "status" in cmd: return FakeResult(0, json.dumps( {"status": "completed", "text": "ok-status"}), "") return FakeResult(0, "", "") return FakeResult(0, "", "") def _default_api(self, socket_path, method, path, payload=None): self.api_calls.append((method, path, payload)) return (200, {}) def _default_launcher(self, cmd): self.launches.append(list(cmd)) self.proc = FakeProc(alive=True) return self.proc def _task(self, task_id="abc123"): return {"id": task_id, "prompt": "做一个加法函数", "project_id": "proj1"} def _audit_events(self): return [e for e, _t, _d, _k in self.audit] def _wake_ok(self, ex, task=None): return ex.wake_vm(task or self._task()) def _known_hosts(self, task_id="abc123"): vm_dir = os.path.join(self.state_dir, "chv-" + task_id) os.makedirs(vm_dir, exist_ok=True) path = os.path.join(vm_dir, "known_hosts") with open(path, "w") as fh: fh.write("127.0.0.1 ssh-ed25519 AAAAC3\n") return path # -- 唤醒失败 ------------------------------------------------------------ class WakeFailureTest(CloseupCase): def test_wake_prechecks_secret_keyscan_fail_closed(self): ex = self.make_exec() # 密钥子集的判定名是 SUB2API_API_KEY（见 ssh_transport # 的 _CHV_SECRET_ENV_NAMES），其它名走普通 env 前缀。 with self.assertRaises(ChvFailure) as ctx: ex.wake_vm(self._task(), extra_env={"SUB2API_API_KEY": "secret-value"}) self.assertEqual(ctx.exception.kind, "secret_env_requires_provisioned") self.assertIn("vm_failed", self._audit_events()) self.assertEqual(self.launches, []) def test_wake_vm_create_failed_closes_vm(self): def api(socket_path, method, path, payload=None): self.api_calls.append((method, path)) if method == "PUT" and path == "/vm.create": return (500, {}) return (200, {}) ex = self.make_exec(api_client=api) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm(self._task()) self.assertEqual(ctx.exception.kind, "vm_create_failed") self.assertIn("vm_failed", self._audit_events()) self.assertEqual(ex._vms, {}) self.assertTrue(any(c[-1] == "shutdown" for c in self.calls)) def test_wake_vm_boot_failed_closes_vm(self): def api(socket_path, method, path, payload=None): self.api_calls.append((method, path)) if method == "PUT" and path == "/vm.boot": return (500, {}) return (200, {}) ex = self.make_exec(api_client=api) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm(self._task()) self.assertEqual(ctx.exception.kind, "vm_boot_failed") self.assertIn("vm_failed", self._audit_events()) self.assertEqual(ex._vms, {}) def test_wake_api_timeout(self): def api(socket_path, method, path, payload=None): return (404, {}) ex = self.make_exec(api_client=api, api_timeout=2) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm(self._task()) self.assertEqual(ctx.exception.kind, "api_timeout") self.assertIn("vm_failed", self._audit_events()) def test_wake_failure_stop_exception_swallowed(self): def api(socket_path, method, path, payload=None): if method == "PUT" and path == "/vm.create": return (500, {}) return (200, {}) def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[-1] == "shutdown": raise RuntimeError("shutdown boom") return FakeResult(0, "", "") ex = self.make_exec(api_client=api, runner=runner) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm(self._task()) # 原始失败仍是 vm_create_failed，不被 stop 的异常掩盖。 self.assertEqual(ctx.exception.kind, "vm_create_failed") def test_wait_guest_ready_host_key_mismatch(self): self._known_hosts() ex = self.make_exec() def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh" and cmd[-1] == "true": return FakeResult(1, "", "Host key verification failed") return FakeResult(0, "", "") ex.runner = runner with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "host_key_mismatch") self.assertIn("host_key_mismatch", self._audit_events()) def test_wait_guest_ready_timeout(self): self._known_hosts() ex = self.make_exec(ready_timeout=2, ready_interval=1) def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh": return FakeResult(1, "", "") return FakeResult(0, "", "") ex.runner = runner with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "guest_not_ready") def test_wait_guest_ready_provisioned_missing_known_hosts(self): ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "host_key_not_pinned") # -- 恢复与快照失败 ------------------------------------------------------ class RestoreSnapshotTest(CloseupCase): def _snap_url(self, name="snap1.raw"): path = os.path.join(self.state_dir, name) os.makedirs(self.state_dir, exist_ok=True) with open(path, "wb") as fh: fh.write(b"S" * 60) return "file://" + path, path def test_restore_failure_closes_vm(self): url, _ = self._snap_url() ex = self.make_exec() def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == self.ch_remote and "restore" in cmd: return FakeResult(1, "", "") return self._default_runner(cmd, timeout) ex.runner = runner with self.assertRaises(RuntimeError): ex.restore_vm(self._task(), url) self.assertEqual(ex._vms, {}) self.assertTrue(any(c[-1] == "shutdown" for c in self.calls)) def test_restore_success_and_wake_from_hibernate(self): url, _ = self._snap_url() ex = self.make_exec() out = ex.restore_vm(self._task(), url) self.assertEqual(out["mode"], "restored") self.assertEqual(out["state"], "running") out2 = ex.wake_from_hibernate(self._task(), url) self.assertEqual(out2["mode"], "restored") def test_restore_provisioned_missing_known_hosts(self): url, _ = self._snap_url() ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) with self.assertRaises(ChvFailure) as ctx: ex.restore_vm(self._task(), url) self.assertEqual(ctx.exception.kind, "host_key_not_pinned") self.assertEqual(self.launches, []) def test_snapshot_not_persisted_missing_and_empty(self): ex = self.make_exec() missing = "file://" + os.path.join(self.state_dir, "nope.raw") os.makedirs(self.state_dir, exist_ok=True) with self.assertRaises(ChvFailure) as ctx: ex.snapshot_vm(self._task(), missing) self.assertEqual(ctx.exception.kind, "snapshot_not_persisted") empty_path = os.path.join(self.state_dir, "empty.raw") open(empty_path, "wb").close() with self.assertRaises(ChvFailure) as ctx: ex.snapshot_vm(self._task(), "file://" + empty_path) self.assertEqual(ctx.exception.kind, "snapshot_not_persisted") def test_snapshot_prunes_oldest_on_total_cap(self): vm = "chv-abc123" vm_dir = os.path.join(self.state_dir, vm) os.makedirs(vm_dir, exist_ok=True) old_path = os.path.join(self.state_dir, "old.raw") with open(old_path, "wb") as fh: fh.write(b"O" * 100) import time with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh: json.dump([{"url": "file://" + old_path, "path": old_path, "size": 100, "created_at": time.time()}], fh) # 上限 150 字节：预检时旧 100 字节通过，新 60 字节记入后 # 总量 160 超帽，最旧一条被剪枝并审计。 cap_gb = 150 / (1024 ** 3) ex = self.make_exec(env_extra={ "CHV_SNAPSHOT_MAX_TOTAL_GB": repr(cap_gb)}) new_path = os.path.join(self.state_dir, "new.raw") with open(new_path, "wb") as fh: fh.write(b"N" * 60) out = ex.snapshot_vm(self._task(), "file://" + new_path) self.assertEqual(out, "file://" + new_path) self.assertFalse(os.path.exists(old_path)) self.assertIn("vm_snapshot_pruned", self._audit_events()) def test_snapshot_quota_count(self): vm_dir = os.path.join(self.state_dir, "chv-abc123") os.makedirs(vm_dir, exist_ok=True) import time with open(os.path.join(vm_dir, "snapshots.json"), "w") as fh: json.dump([{"url": "file://x", "path": None, "size": 0, "created_at": time.time()}], fh) ex = self.make_exec(env_extra={"CHV_SNAPSHOT_MAX_COUNT": "1"}) url, _ = self._snap_url("snap2.raw") with self.assertRaises(ChvFailure) as ctx: ex.snapshot_vm(self._task(), url) self.assertEqual(ctx.exception.kind, "snapshot_quota_exceeded") def test_hibernate_snapshot_failure_does_not_shutdown(self): ex = self.make_exec() with self.assertRaises(ValueError): ex.hibernate_vm(self._task(), "") self.assertFalse(any(c[-1] == "shutdown" for c in self.calls)) # -- 取消与清理 ---------------------------------------------------------- class CancelCleanupTest(CloseupCase): def test_cancel_vm_really_shuts_down(self): ex = self.make_exec() self._wake_ok(ex) proc = self.proc self.assertTrue(ex.cancel_vm(self._task())) self.assertTrue(proc.terminated) self.assertEqual(ex._vms, {}) self.assertIn("vm_shutdown", self._audit_events()) self.assertIn("vm_destroy", self._audit_events()) def test_shutdown_vm_socket_residual(self): ex = self.make_exec() os.makedirs(self.state_dir, exist_ok=True) sock = os.path.join(self.state_dir, "chv-abc123.sock") with open(sock, "w") as fh: fh.write("") self.assertTrue(ex.shutdown_vm(self._task())) self.assertFalse(os.path.exists(sock)) def test_recycle_vm_success_and_missing_id(self): ex = self.make_exec() self._wake_ok(ex) out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "recycled") self.assertEqual(out["preserved"], ["task_checkpoints", "task_attempts"]) self.assertIn("vm_recycled", self._audit_events()) with self.assertRaises(ValueError): ex.recycle_vm({}) def test_cleanup_stale_vms_dead_socket_and_live_skip(self): ex = self.make_exec() os.makedirs(self.state_dir, exist_ok=True) for name in ("chv-dead.sock", "chv-live.sock", "other.sock"): with open(os.path.join(self.state_dir, name), "w") as fh: fh.write("") ex._vms["chv-live"] = FakeProc(alive=True) cleaned = ex.cleanup_stale_vms() self.assertEqual(cleaned, ["chv-dead"]) self.assertFalse( os.path.exists(os.path.join(self.state_dir, "chv-dead.sock"))) self.assertTrue( os.path.exists(os.path.join(self.state_dir, "other.sock"))) def test_cleanup_stale_vms_purges_dead_registry(self): ex = self.make_exec() os.makedirs(self.state_dir, exist_ok=True) ex._vms["chv-ghost"] = FakeProc(alive=False) self.assertEqual(ex.cleanup_stale_vms(), []) self.assertEqual(ex._vms, {}) def test_cleanup_stale_vms_missing_state_dir(self): ex = self.make_exec() self.assertFalse(os.path.exists(self.state_dir)) self.assertEqual(ex.cleanup_stale_vms(), []) def test_gc_stale_disks_deletes_stale_keeps_live(self): ex = self.make_exec() stale_dir = os.path.join(self.state_dir, "chv-stale") live_dir = os.path.join(self.state_dir, "chv-live") os.makedirs(stale_dir, exist_ok=True) os.makedirs(live_dir, exist_ok=True) with open(os.path.join(stale_dir, "disk.raw"), "wb") as fh: fh.write(b"D") with open(os.path.join(live_dir, "disk.raw"), "wb") as fh: fh.write(b"D") with open(os.path.join(self.state_dir, "chv-live.sock"), "w") as fh: fh.write("") out = ex.gc_stale_disks() self.assertEqual(out["deleted"], ["chv-stale"]) self.assertFalse( os.path.exists(os.path.join(stale_dir, "disk.raw"))) self.assertTrue(os.path.exists(os.path.join(live_dir, "disk.raw"))) def test_cleanup_stale_secrets_live_keeps_seed_dead_removes(self): ex = self.make_exec() live_dir = os.path.join(self.state_dir, "chv-live") dead_dir = os.path.join(self.state_dir, "chv-dead") os.makedirs(live_dir, exist_ok=True) os.makedirs(dead_dir, exist_ok=True) for vm_dir in (live_dir, dead_dir): with open(os.path.join(vm_dir, "ssh_host_ed25519_key"), "w") as fh: fh.write("priv") with open(os.path.join(vm_dir, "seed.iso"), "wb") as fh: fh.write(b"S") with open(os.path.join(self.state_dir, "chv-live.sock"), "w") as fh: fh.write("") cleaned = ex.cleanup_stale_secrets() self.assertEqual(sorted(cleaned), ["chv-dead", "chv-live"]) self.assertFalse( os.path.exists(os.path.join(live_dir, "ssh_host_ed25519_key"))) self.assertTrue(os.path.exists(os.path.join(live_dir, "seed.iso"))) self.assertFalse( os.path.exists(os.path.join(dead_dir, "ssh_host_ed25519_key"))) self.assertFalse(os.path.exists(os.path.join(dead_dir, "seed.iso"))) def test_stop_dirty_marker_on_cleanup_failure(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-abc123") seed_dir = os.path.join(vm_dir, "seed") os.makedirs(seed_dir, exist_ok=True) with open(os.path.join(seed_dir, "user-data"), "w") as fh: fh.write("seed") with open(os.path.join(self.state_dir, "chv-abc123.sock"), "w") as fh: fh.write("") # 本环境以 root 运行，chmod 000/0555 会被 root 绕过， # 权限失败分支改用 rmtree 抛错模拟（只包 stop_vm 一次 # 调用，上下文退出即恢复，不影响 tearDown 清理）。 import shutil real_rmtree = shutil.rmtree def _failing_rmtree(path, *args, **kwargs): if path == seed_dir: raise OSError("mocked rmtree failure") return real_rmtree(path, *args, **kwargs) with mock.patch("shutil.rmtree", side_effect=_failing_rmtree): self.assertTrue(ex.stop_vm(self._task())) self.assertIn("vm_key_cleanup_failed", self._audit_events()) marker = dirty_marker_path(self.state_dir, "chv-abc123") self.assertTrue(os.path.exists(marker)) def test_stop_clears_dirty_marker_on_success(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-abc123") os.makedirs(vm_dir, exist_ok=True) marker = dirty_marker_path(self.state_dir, "chv-abc123") os.makedirs(os.path.dirname(marker), exist_ok=True) with open(marker, "w") as fh: fh.write("cleanup failed\n") with open(os.path.join(self.state_dir, "chv-abc123.sock"), "w") as fh: fh.write("") self.assertTrue(ex.stop_vm(self._task())) self.assertFalse(os.path.exists(marker)) # -- 中途死亡分类与只读 execute ------------------------------------------ class VmDeathTest(CloseupCase): def _wake_with_dead_on_dispatch(self, mode): ex = self.make_exec() def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh" and "dispatch" in cmd: proc = ex._vms.get("chv-abc123") if proc is not None: proc._alive = False if mode == "raise": raise RuntimeError("ssh broken") return FakeResult(0, json.dumps( {"status": "failed", "text": "boom"}), "") return self._default_runner(cmd, timeout) ex.runner = runner return ex def test_run_with_vm_exception_vm_dead_interrupted(self): ex = self._wake_with_dead_on_dispatch("raise") out = ex.dispatch_task(self._task()) self.assertEqual(out["status"], "interrupted") self.assertTrue(out["vm_dead"]) def test_run_with_vm_failed_status_vm_dead_reclassified(self): ex = self._wake_with_dead_on_dispatch("failed") out = ex.dispatch_task(self._task()) self.assertEqual(out["status"], "interrupted") self.assertTrue(out["vm_dead"]) self.assertIn("VM 中途死亡", out["interrupted_reason"]) def test_run_with_vm_exception_vm_alive_reraises(self): ex = self.make_exec() def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh" and "dispatch" in cmd: raise RuntimeError("ssh broken") return self._default_runner(cmd, timeout) ex.runner = runner with self.assertRaises(RuntimeError): ex.dispatch_task(self._task()) def test_execute_dead_vm_reclassified(self): ex = self.make_exec() ex._vms["chv-abc123"] = FakeProc(alive=False) os.makedirs(self.state_dir, exist_ok=True) with open(os.path.join(self.state_dir, "chv-abc123.sock"), "w") as fh: fh.write("") def runner(cmd, timeout=None): self.calls.append(list(cmd)) raise RuntimeError("ssh broken") ex.runner = runner with self.assertRaises(ChvFailure) as ctx: ex.execute(self._task()) self.assertEqual(ctx.exception.kind, "vm_not_running") def test_execute_not_running_no_runner_call(self): ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.execute(self._task()) self.assertEqual(ctx.exception.kind, "vm_not_running") self.assertEqual(self.calls, []) # -- 入口校验 ------------------------------------------------------------ class EntryGuardTest(CloseupCase): def test_dispatch_run_goal_guards(self): ex = self.make_exec() with self.assertRaises(ValueError): ex.dispatch_task({}) with self.assertRaises(ValueError): ex.dispatch_task({"id": "x"}) with self.assertRaises(ValueError): ex.run_task({"id": "x"}) with self.assertRaises(ValueError): ex.goal_task({"id": "x", "goal": "g", "goal_max_rounds": 101}) with self.assertRaises(ValueError): ex.goal_task({"id": "x", "goal": "g", "verify_command": "v", "goal_max_rounds": 0}) with self.assertRaises(ValueError): ex.goal_resume_task({}) with self.assertRaises(ValueError): ex.goal_resume_breakpoint_task({}) with self.assertRaises(ValueError): ex.wake_vm({}) def test_dispatch_wake_failure_does_not_run_guest(self): ex = self.make_exec(binary="/nonexistent/cloud-hypervisor") with self.assertRaises(ChvFailure): ex.dispatch_task(self._task()) self.assertFalse(any("dispatch" in c for c in self.calls)) # -- preflight 事后改写校验 ------------------------------------------------ class PreflightMutationTest(CloseupCase): def test_reuse_blocked_and_tenant_undeclared(self): ex = self.make_exec() ex.disk_mode = "reuse" ex.reuse_experiment = False with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "disk_mode_reuse_blocked") ex.reuse_experiment = True ex.single_tenant_project = None with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "reuse_tenant_undeclared") def test_config_and_numeric_mutations(self): cases = [ ("cpus", "not-an-int"), ("memory", "bogus-mem"), ("ssh_port", 99999), ("max_vms", -1), ("disk_gb", 0), ("api_timeout", 0), ("warm_pool_size", -1), ("snapshot_max_count", -1), ("host_key_mode", "bogus-mode"), ] for attr, value in cases: with self.subTest(attr=attr): ex = self.make_exec() setattr(ex, attr, value) with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "config_invalid") def test_cpu_flag_missing_and_kvm_unreadable(self): with open(self.cpuinfo, "w") as fh: fh.write("flags : fpu sse\n") ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "cpu_flag_missing") with open(self.cpuinfo, "w") as fh: fh.write("flags : fpu vmx sse\n") ex = self.make_exec() # 本环境以 root 运行，chmod 000 会被 root 绕过， # kvm 不可读/写分支改用 os.access 模拟（只包本次 # preflight，上下文退出即恢复）。 real_access = os.access def _fake_access(path, mode, *args, **kwargs): if path == self.kvm and mode in (os.R_OK, os.W_OK): return False return real_access(path, mode, *args, **kwargs) with mock.patch("os.access", side_effect=_fake_access): with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "kvm_missing") def test_net_dual_warns_but_passes(self): ex = self.make_exec(env_extra={ "CHV_NET": "tap0", "CHV_NET_JSON": '[{"tap": "tap0"}]'}) self.assertTrue(ex.preflight()) self.assertIn("net_config_dual", self._audit_events()) def test_net_json_invalid(self): ex = self.make_exec(env_extra={"CHV_NET_JSON": "{not json"}) with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "net_config_invalid") # -- worker_adapter 剩余分支 ------------------------------------------------ class WorkerAdapterCloseupTest(unittest.TestCase): def test_call_breakpoint_extra_env_forwarded(self): seen = {} class _Ex: def goal_resume_breakpoint_task(self, task, model_provider=None, extra_env=None): seen["model_provider"] = model_provider seen["extra_env"] = extra_env return "ok" out = call_goal_resume_breakpoint( _Ex(), {"id": "x"}, model_provider="m", extra_env={"A": "1"}) self.assertEqual(out, "ok") self.assertEqual(seen["extra_env"], {"A": "1"}) out = call_goal_resume_breakpoint(_Ex(), {"id": "x"}) self.assertEqual(seen["extra_env"], None) def test_call_breakpoint_signature_error_fallback(self): class _BadSig: __signature__ = "not-a-signature" def __call__(self, task): return "fallback-ok" class _Ex: def __init__(self): self.goal_resume_breakpoint_task = _BadSig() self.assertEqual( call_goal_resume_breakpoint(_Ex(), {"id": "x"}), "fallback-ok") def test_reconcile_non_chv_and_failure_audited(self): self.assertEqual(reconcile_warm_pool(object(), lambda *a: None, "when"), []) audit = [] # CHV 失败分支：patch orphans 调和函数抛错，执行器 # 用 spec Mock 并把 __class__ 设为真执行器类，使 # isinstance 判定进入 CHV 分支。 from cloud.chv.executor import CloudHypervisorExecutor as _RealEx self.assertTrue(inspect.isclass(_RealEx)) with mock.patch( "cloud.chv.orphans.reconcile_orphans_for_executor", side_effect=RuntimeError("boom")): fake_ex = mock.Mock(spec=_RealEx) # spec Mock 的 isinstance 判定走 __class__，需显式 # 把类设为真执行器类才能进 CHV 分支。 fake_ex.__class__ = _RealEx out = reconcile_warm_pool( fake_ex, lambda *a: audit.append(a), "when") self.assertEqual(out, []) self.assertTrue(any("reconcile_failed" in a[1] for a in audit)) def test_cancel_watchdog_cancel_vm_none(self): stop = threading.Event() requested = {"t1"} cancel_watchdog_loop("t1", "chv-t1", stop, requested, lambda tid: None, None, object()) # 命中取消但 cancel_vm 为 None 时不炸、直接退出。 def test_quota_watchdog_not_over_then_stop(self): stop = threading.Event() def check(): stop.set() return False calls = [] quota_watchdog_loop("t1", "chv-t1", stop, check, lambda tid: calls.append(tid), None, object(), interval=0.01) self.assertEqual(calls, []) def test_watchdog_vm_name_non_callable(self): self.assertIsNone(watchdog_vm_name(object(), {"id": "x"})) if __name__ == "__main__": unittest.main() # -- 收尾第二批：防御性 except 与文件/入口变体 -------------------------- class RaisingProc: def poll(self): raise RuntimeError("poll boom") def terminate(self): raise RuntimeError("terminate boom") def wait(self, timeout=None): raise RuntimeError("wait boom") def kill(self): raise RuntimeError("kill boom") class DefensiveBranchTest(CloseupCase): def test_default_popen_launcher_runs_true(self): from cloud.chv.executor import _default_popen_launcher proc = _default_popen_launcher(["true"]) self.assertEqual(proc.wait(), 0) def test_binary_ok_bare_name_via_which(self): ex = self.make_exec(binary="true", ch_remote="true") self.assertTrue(ex.preflight()) def test_kill_proc_none_and_raising(self): ex = self.make_exec() ex._kill_proc(None) ex._kill_proc(RaisingProc()) # terminate/wait/kill 全抛也不炸 def test_active_and_persistent_count_poll_exception(self): ex = self.make_exec() self.assertEqual(ex._active_count(), 0) ex._vms["chv-bad"] = RaisingProc() self.assertEqual(ex._active_count(), 1) self.assertEqual(ex._persistent_vm_count(), 0) def test_preflight_cpuinfo_missing_and_unset_files(self): ex = self.make_exec(cpuinfo_path="/nonexistent/cpuinfo") with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "cpu_flag_missing") for kwargs in ({"kernel": None}, {"rootfs": None}, {"ssh_key": None}): with self.subTest(kwargs=kwargs): ex = self.make_exec(**kwargs) with self.assertRaises(ChvFailure): ex.preflight() def test_preflight_unreadable_files_via_access_mock(self): ex = self.make_exec() real_access = os.access def _fake_access(path, mode, *args, **kwargs): if path in (self.kernel, self.rootfs, self.ssh_key) \ and mode == os.R_OK: return False return real_access(path, mode, *args, **kwargs) with mock.patch("os.access", side_effect=_fake_access): with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertIn(ctx.exception.kind, ("kernel_missing", "image_missing", "ssh_key_missing")) def test_preflight_state_dir_is_file(self): blocker = os.path.join(self.dir, "blocker") with open(blocker, "w") as fh: fh.write("x") ex = self.make_exec(state_dir=blocker) with self.assertRaises(ChvFailure) as ctx: ex.preflight() self.assertEqual(ctx.exception.kind, "state_dir_insecure") def test_build_vm_command_with_net_and_config_errors(self): ex = self.make_exec(net="tap0") cmd = ex._build_vm_command(self._task()) self.assertIn("--net", cmd) self.assertIn("tap0", cmd) ex.cpus = "bogus" with self.assertRaises(ValueError): ex.build_vm_config() ex.cpus = 2 ex.net_json = "{bad" with self.assertRaises(ValueError): ex.build_vm_config() def test_net_guest_ip_invalid_and_ssh_target_undeclared(self): ex = self.make_exec() ex.net_json = "{bad" self.assertIsNone(ex._net_guest_ip()) ex2 = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"}) with self.assertRaises(ChvFailure) as ctx: ex2.resolve_ssh_host() self.assertEqual(ctx.exception.kind, "ssh_target_undeclared") def test_capture_host_key_resolve_and_runner_and_write_failures(self): ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"}) self.assertFalse(ex.capture_host_key(self._task())) ex = self.make_exec() def raising_runner(cmd, timeout=None): raise RuntimeError("keyscan boom") ex.runner = raising_runner self.assertFalse(ex.capture_host_key(self._task())) ex = self.make_exec() blocker = os.path.join(self.state_dir, "chv-abc123") os.makedirs(self.state_dir, exist_ok=True) with open(blocker, "w") as fh: fh.write("x") def keyscan_runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh-keyscan": return FakeResult(0, "[127.0.0.1]:2222 ssh-ed25519 AAAA-good", "") return FakeResult(0, "", "") ex.runner = keyscan_runner self.assertFalse(ex.capture_host_key(self._task())) def test_build_ssh_command_fallback_and_secret_not_delivered(self): ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0"}) cmd = ex.build_ssh_command(self._task(), ["true"]) self.assertIn("127.0.0.1", " ".join(cmd)) ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.build_ssh_command(self._task(), ["true"], extra_env={"SUB2API_API_KEY": "v"}) self.assertEqual(ctx.exception.kind, "secret_env_not_delivered") def test_write_provisioned_known_hosts_fallback(self): ex = self.make_exec(env_extra={"CHV_SSH_FORWARD": "0", "CHV_HOST_KEY_MODE": "provisioned"}) vm_dir = os.path.join(self.state_dir, "chv-abc123") os.makedirs(vm_dir, exist_ok=True) paths = {"known_hosts": os.path.join(vm_dir, "known_hosts")} ex._write_provisioned_known_hosts(paths, "ssh-ed25519 AAAA") with open(paths["known_hosts"]) as fh: self.assertIn("127.0.0.1", fh.read()) def test_provision_host_key_dirty_refused(self): ex = self.make_exec() marker = dirty_marker_path(self.state_dir, "chv-abc123") os.makedirs(os.path.dirname(marker), exist_ok=True) with open(marker, "w") as fh: fh.write("cleanup failed\n") with self.assertRaises(ChvFailure) as ctx: ex.provision_host_key(self._task()) self.assertEqual(ctx.exception.kind, "vm_dirty_refused") def test_api_call_raises_then_api_timeout(self): def api(socket_path, method, path, payload=None): raise ConnectionError("socket gone") ex = self.make_exec(api_client=api, api_timeout=2) with self.assertRaises(ChvFailure) as ctx: ex.wait_api_ready(self._task()) self.assertEqual(ctx.exception.kind, "api_timeout") def test_wait_guest_runner_exception_retried_then_timeout(self): self._known_hosts() ex = self.make_exec(ready_timeout=2, ready_interval=1) def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh": raise RuntimeError("ssh boom") return FakeResult(0, "", "") ex.runner = runner with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "guest_not_ready") def test_stop_exceptions_swallowed_around_stop_vm(self): # wait_guest_ready 的 provisioned 守卫、mismatch 分支、 # wake_vm 与 restore_vm 的 stop_vm 外层 except：stop_vm # 本身不抛，改用 mock 让它抛，验证原始失败不被掩盖。 ex = self.make_exec(env_extra={"CHV_HOST_KEY_MODE": "provisioned"}) ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom")) with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "host_key_not_pinned") def api(socket_path, method, path, payload=None): if method == "PUT" and path == "/vm.create": return (500, {}) return (200, {}) ex = self.make_exec(api_client=api) ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom")) with self.assertRaises(ChvFailure) as ctx: ex.wake_vm(self._task()) self.assertEqual(ctx.exception.kind, "vm_create_failed") url, _ = self._snap_url_for(ex) ex = self.make_exec() ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom")) def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == self.ch_remote and "restore" in cmd: return FakeResult(1, "", "") return self._default_runner(cmd, timeout) ex.runner = runner with self.assertRaises(RuntimeError): ex.restore_vm(self._task(), url) def _snap_url_for(self, ex): path = os.path.join(self.state_dir, "snapx.raw") os.makedirs(self.state_dir, exist_ok=True) with open(path, "wb") as fh: fh.write(b"S" * 10) return "file://" + path, path def test_registered_net_entries_variants(self): ex = self.make_exec() os.makedirs(os.path.join(self.state_dir, "chv-skip"), exist_ok=True) with open(os.path.join(self.state_dir, "chv-skip", "net.json"), "w") as fh: json.dump([{"tap": "tap9", "ip": "10.0.0.9"}], fh) os.makedirs(os.path.join(self.state_dir, "chv-bad"), exist_ok=True) with open(os.path.join(self.state_dir, "chv-bad", "net.json"), "w") as fh: fh.write('{"not": "a list"}') os.makedirs(os.path.join(self.state_dir, "chv-empty"), exist_ok=True) out = ex._registered_net_entries(exclude_vm="chv-skip") self.assertEqual(out, []) ex2 = self.make_exec(state_dir="/nonexistent-state-dir-x") self.assertEqual(ex2._registered_net_entries(), []) def test_socket_permissions_stat_failure(self): ex = self.make_exec() sock = os.path.join(self.state_dir, "chv-abc123.sock") os.makedirs(self.state_dir, exist_ok=True) with open(sock, "w") as fh: fh.write("") real_stat = os.stat with mock.patch("os.stat", side_effect=OSError("stat boom")): ex._check_api_socket_permissions(sock) # 不抛，只跳过 self.assertTrue(real_stat(sock) is not None) def test_rewake_kills_old_proc_and_start_vm_alias(self): ex = self.make_exec() self._wake_ok(ex) old = self.proc out = ex.start_vm(self._task()) self.assertEqual(out["mode"], "cold") self.assertTrue(old.terminated) def test_validate_restore_guards(self): ex = self.make_exec() with self.assertRaises(ValueError): ex._validate_restore({}, "file://x") with self.assertRaises(ValueError): ex._validate_restore(self._task(), "") with self.assertRaises(ChvFailure) as ctx: ex._validate_restore(self._task(), "snap.raw") self.assertEqual(ctx.exception.kind, "snapshot_path_not_allowed") def test_stop_registered_proc_poll_exception(self): ex = self.make_exec() ex._vms["chv-bad"] = RaisingProc() ex._stop_registered_proc("chv-bad", RaisingProc()) self.assertEqual(ex._vms, {}) def test_cleanup_net_remove_failure_marks_dirty(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-abc123") os.makedirs(vm_dir, exist_ok=True) os.makedirs(os.path.join(vm_dir, "net.json"), exist_ok=True) with open(os.path.join(self.state_dir, "chv-abc123.sock"), "w") as fh: fh.write("") self.assertTrue(ex.stop_vm(self._task())) marker = dirty_marker_path(self.state_dir, "chv-abc123") self.assertTrue(os.path.exists(marker)) def test_write_dirty_marker_failure_swallowed(self): blocker = os.path.join(self.dir, "blocker2") with open(blocker, "w") as fh: fh.write("x") ex = self.make_exec(state_dir=blocker) ex._write_dirty_marker("chv-abc123", True) # 不抛 def test_stop_socket_remove_failure_swallowed(self): ex = self.make_exec() os.makedirs(os.path.join(self.state_dir, "chv-abc123.sock"), exist_ok=True) self.assertTrue(ex.stop_vm(self._task())) def test_gc_poll_exception_and_delete_failure(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-bad") os.makedirs(vm_dir, exist_ok=True) with open(os.path.join(vm_dir, "disk.raw"), "wb") as fh: fh.write(b"D") ex._vms["chv-bad"] = RaisingProc() out = ex.gc_stale_disks() self.assertEqual(out["deleted"], ["chv-bad"]) vm2 = os.path.join(self.state_dir, "chv-dir") os.makedirs(os.path.join(vm2, "disk.raw"), exist_ok=True) out = ex.gc_stale_disks() self.assertEqual(out["failed"], ["chv-dir"]) self.assertIn("vm_disk_delete_failed", self._audit_events()) def test_is_live_and_proc_dead_poll_exception(self): ex = self.make_exec() ex._vms["chv-bad"] = RaisingProc() self.assertFalse(ex._is_live_vm("chv-bad")) self.assertFalse(ex._vm_proc_dead("chv-bad")) self.assertFalse(ex._vm_is_running("chv-bad")) def test_clean_one_vm_secrets_remove_failures(self): ex = self.make_exec() vm_dir = os.path.join(self.state_dir, "chv-x") os.makedirs(os.path.join(vm_dir, "ssh_host_ed25519_key"), exist_ok=True) removed_priv, removed_seed = ex._clean_one_vm_secrets( "chv-x", vm_dir) self.assertFalse(removed_priv) vm2 = os.path.join(self.state_dir, "chv-y") os.makedirs(os.path.join(vm2, "seed"), exist_ok=True) import shutil real_rmtree = shutil.rmtree with mock.patch("shutil.rmtree", side_effect=OSError("rmtree boom")): removed_priv, removed_seed = ex._clean_one_vm_secrets( "chv-y", vm2) self.assertFalse(removed_seed) self.assertTrue(real_rmtree is not None) def test_cleanup_stale_vms_subcall_failures_and_poll_branches(self): ex = self.make_exec() os.makedirs(self.state_dir, exist_ok=True) with open(os.path.join(self.state_dir, "notes.txt"), "w") as fh: fh.write("not a socket") with open(os.path.join(self.state_dir, "chv-dead.sock"), "w") as fh: fh.write("") ex._vms["chv-dead"] = RaisingProc() ex.cleanup_stale_secrets = mock.Mock( side_effect=RuntimeError("secrets boom")) ex.gc_stale_disks = mock.Mock( side_effect=RuntimeError("gc boom")) cleaned = ex.cleanup_stale_vms() self.assertEqual(cleaned, ["chv-dead"]) ex2 = self.make_exec() ex2._vms["chv-ghost"] = RaisingProc() ex2._purge_dead_registry() self.assertEqual(ex2._vms, {}) def test_run_with_vm_finally_shutdown_exception_swallowed(self): ex = self.make_exec() real_shutdown = ex.shutdown_vm ex.shutdown_vm = mock.Mock(side_effect=RuntimeError("sd boom")) # _run_with_vm 的 finally 里 shutdown 抛错应被吞掉， # dispatch 仍返回 completed（finally 不改返回值）。 out = ex.dispatch_task(self._task()) self.assertEqual(out["status"], "completed") ex.shutdown_vm = real_shutdown def test_execute_guards_and_alive_reraise(self): ex = self.make_exec() with self.assertRaises(ValueError): ex.execute({}) ex._vms["chv-abc123"] = FakeProc(alive=True) os.makedirs(self.state_dir, exist_ok=True) with open(os.path.join(self.state_dir, "chv-abc123.sock"), "w") as fh: fh.write("") def runner(cmd, timeout=None): raise RuntimeError("ssh boom") ex.runner = runner with self.assertRaises(RuntimeError): ex.execute(self._task()) def test_entry_success_branches(self): ex = self.make_exec() out = ex.dispatch_task({**self._task(), "repo": "https://example.com/r.git", "plan": True}) self.assertEqual(out["status"], "completed") ssh_cmds = [c for c in self.calls if c and c[0] == "ssh"] self.assertTrue(any("https://example.com/r.git" in " ".join(c) for c in ssh_cmds)) self.calls.clear() out = ex.run_task(self._task()) self.assertEqual(out["status"], "completed") with self.assertRaises(ValueError): ex.run_task({}) with self.assertRaises(ValueError): ex.goal_task({"id": "x"}) with self.assertRaises(ValueError): ex.goal_task({}) out = ex.goal_task({**self._task(), "goal": "g", "verify_command": "make check", "repo": "https://example.com/r.git"}) self.assertEqual(out["status"], "completed") goal_cmds = [c for c in self.calls if c and c[0] == "ssh" and "goal" in c] self.assertTrue(any("make check" in " ".join(c) for c in goal_cmds)) self.calls.clear() out = ex.goal_resume_task(self._task(), goal_text="new goal", max_rounds=2) self.assertEqual(out["status"], "completed") resume_cmds = [c for c in self.calls if c and c[0] == "ssh" and "goal-resume" in c] self.assertTrue(any("--goal-text" in c for c in resume_cmds)) self.calls.clear() out = ex.goal_resume_task(self._task(), max_rounds=5) self.assertEqual(out["status"], "completed") resume_cmds = [c for c in self.calls if c and c[0] == "ssh" and "goal-resume" in c] self.assertTrue(any("--max-rounds" in c for c in resume_cmds)) class WorkerAdapterCloseup2Test(unittest.TestCase): def test_is_chv_executor(self): from cloud.chv.worker_adapter import is_chv_executor self.assertFalse(is_chv_executor(object())) def test_reconcile_audit_fn_failures_swallowed(self): from cloud.chv.executor import CloudHypervisorExecutor as _RealEx from cloud.chv.worker_adapter import reconcile_warm_pool with mock.patch( "cloud.chv.orphans.reconcile_orphans_for_executor", side_effect=RuntimeError("boom")): fake_ex = mock.Mock(spec=_RealEx) fake_ex.__class__ = _RealEx def raising_audit(*args): raise RuntimeError("audit boom") out = reconcile_warm_pool(fake_ex, raising_audit, "when") self.assertEqual(out, []) with mock.patch( "cloud.chv.orphans.reconcile_orphans_for_executor", return_value=["chv-x"]): fake_ex = mock.Mock(spec=_RealEx) fake_ex.__class__ = _RealEx out = reconcile_warm_pool(fake_ex, raising_audit, "when") self.assertEqual(out, ["chv-x"]) def test_cancel_watchdog_get_task_variants(self): from cloud.chv.worker_adapter import cancel_watchdog_loop stop = threading.Event() def get_task_raising(tid): stop.set() raise RuntimeError("db boom") cancel_watchdog_loop("t1", "chv-t1", stop, set(), get_task_raising, None, object()) stop2 = threading.Event() def get_task_running(tid): stop2.set() return {"status": "running"} cancel_watchdog_loop("t1", "chv-t1", stop2, set(), get_task_running, None, object()) def test_cancel_watchdog_cancel_vm_raises_swallowed(self): from cloud.chv.worker_adapter import cancel_watchdog_loop stop = threading.Event() def bad_cancel(vm): raise RuntimeError("cancel boom") cancel_watchdog_loop("t1", "chv-t1", stop, {"t1"}, lambda tid: None, bad_cancel, object()) def test_quota_watchdog_check_raises_swallowed(self): from cloud.chv.worker_adapter import quota_watchdog_loop stop = threading.Event() def bad_check(): stop.set() raise RuntimeError("quota boom") quota_watchdog_loop("t1", "chv-t1", stop, bad_check, lambda tid: None, None, object(), interval=0.01) class ExecutorLastLinesTest(CloseupCase): def test_net_guest_ip_no_ip_entry_returns_none(self): ex = self.make_exec() ex.net_json = '[{"tap": "tap0"}]' self.assertIsNone(ex._net_guest_ip()) def test_wait_guest_mismatch_stop_exception_swallowed(self): self._known_hosts() ex = self.make_exec() ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom")) def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh" and cmd[-1] == "true": return FakeResult(1, "", "Host key verification failed") return FakeResult(0, "", "") ex.runner = runner with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "host_key_mismatch") def test_wait_guest_timeout_stop_exception_swallowed(self): self._known_hosts() ex = self.make_exec(ready_timeout=2, ready_interval=1) ex.stop_vm = mock.Mock(side_effect=RuntimeError("stop boom")) def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh": return FakeResult(1, "", "") return FakeResult(0, "", "") ex.runner = runner with self.assertRaises(ChvFailure) as ctx: ex.wait_guest_ready(self._task()) self.assertEqual(ctx.exception.kind, "guest_not_ready") def test_record_net_entries_write_failure_only_warns(self): ex = self.make_exec() blocker = os.path.join(self.state_dir, "chv-x") os.makedirs(self.state_dir, exist_ok=True) with open(blocker, "w") as fh: fh.write("x") with self.assertLogs("cloud.cloud_hypervisor", level="WARNING") as logs: ex._record_net_entries("chv-x", [{"tap": "tap0"}]) self.assertTrue(any("net_registry_write_failed" in line for line in logs.output)) def test_socket_permissions_stat_failure_after_exists(self): ex = self.make_exec() sock = os.path.join(self.state_dir, "chv-abc123.sock") os.makedirs(self.state_dir, exist_ok=True) with open(sock, "w") as fh: fh.write("") real_stat = os.stat calls = {"n": 0} def _stat_then_raise(path, *args, **kwargs): calls["n"] += 1 if calls["n"] == 1: return real_stat(path, *args, **kwargs) raise OSError("stat boom") # 第 1 次 stat 是 exists() 的探查（须成功才进到显式 # stat），第 2 次显式 stat 抛错 → except 分支 return。 with mock.patch("os.stat", side_effect=_stat_then_raise): ex._check_api_socket_permissions(sock) def test_vm_proc_dead_unregistered_returns_false(self): ex = self.make_exec() self.assertFalse(ex._vm_proc_dead(self._task()))