From 8c8d62f70d83943e7f42e858600f6b65a97870cc Mon Sep 17 00:00:00 2001
From: host <host@test.local>
Date: Sun, 1 Nov 2026 07:00:21 +0000
Subject: [PATCH] =?UTF-8?q?test(chv):=20=E8=A1=A5=E9=BD=90=20chv=20?=
 =?UTF-8?q?=E6=A8=A1=E5=9D=97=E5=89=A9=E4=BD=99=E8=A6=86=E7=9B=96=E5=88=B0?=
 =?UTF-8?q?=20100%=EF=BC=88=E7=94=9F=E4=BA=A7=E4=B8=8E=E5=AE=9E=E9=AA=8C?=
 =?UTF-8?q?=E5=88=86=E5=BC=80=E8=AE=A1=E5=88=86=EF=BC=89?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 cloud/tests/test_chv_cover100.py | 618 +++++++++++++++++++++++++++++++
 1 file changed, 618 insertions(+)
 create mode 100644 cloud/tests/test_chv_cover100.py

diff --git a/cloud/tests/test_chv_cover100.py b/cloud/tests/test_chv_cover100.py
new file mode 100644
index 0000000..db7595e
--- /dev/null
+++ b/cloud/tests/test_chv_cover100.py
@@ -0,0 +1,618 @@
+"""CHV 覆盖收尾轨（feat-chv-cover100）新增测试。
+
+只补本轨实测基线（coverage --source=cloud/chv，跑
+test_chv*.py + test_cloud_hypervisor.py：1925 行、缺 80 行、
+96%）里仍未覆盖的行，生产代码零改动。每项都是行为断言：
+
+- api_client：无冒号头行跳过、分片到达的体部、不可解析
+  的状态行、带 payload 的 unix socket 调用、normalize 的
+  dict/对象两分支（此前只覆 tuple 分支）；
+- compat215：commonpath 跨盘符 ValueError 回 None；
+- config：磁盘 GB 非数字、快照数非数字、ready_interval 为 0
+  回落 5 秒（旧语义）；
+- disk：目录统计里文件消失、配额早退、disk_usage 失败与
+  宿主余量不足、已存在磁盘不可读、拷贝失败、剪枝删文件
+  失败、快照总量统计的文件大小回落与失败、file:// 快照
+  文件不存在、总量帽删旧文件失败；
+- hardening：state_dir stat 失败、chmod 后仍宽松、socket
+  死活的空/缺失/异常三态、空 net 配置、非 dict 条目、
+  egress 空值与不可解析 URL；
+- hostkeys：无 runner 时 genisoimage/keygen 走 subprocess
+  的失败路径、seed/key 私钥 chmod 失败、无 runcmd 时密钥
+  环境条目仍追加（不丢）；
+- experimental_pool（单独计分，非生产路径）：registry
+  坏条目计数、非 dict task 落盘、reconcile 跳过 warm/
+  shutdown 失败/_emit 失败、prewarm 重名跳过与唤醒失败、
+  acquire _emit 失败、release shutdown 失败。
+
+全部用假 socket/假 runner/假执行器，绝不连真实网络、
+不起真实 VM、不改生产行为。
+"""
+
+import json
+import os
+import shutil
+import socket
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+from cloud.chv import api_client
+from cloud.chv import compat215
+from cloud.chv import config as chv_config
+from cloud.chv import disk as chv_disk
+from cloud.chv import experimental_pool
+from cloud.chv import hardening
+from cloud.chv import hostkeys
+from cloud.chv.errors import ChvFailure
+from cloud.chv.experimental_pool import WarmPool
+
+
+class _FakeApiSocket:
+    """_unix_socket_api 用的假 socket：按 recv 序列供数据。"""
+
+    def __init__(self, chunks):
+        self._chunks = list(chunks)
+        self.sent = b""
+        self.closed = False
+
+    def settimeout(self, _timeout):
+        pass
+
+    def connect(self, _path):
+        pass
+
+    def sendall(self, data):
+        self.sent += data
+
+    def recv(self, _size):
+        if not self._chunks:
+            return b""
+        return self._chunks.pop(0)
+
+    def close(self):
+        self.closed = True
+
+
+class ApiClientCover100Test(unittest.TestCase):
+    def test_header_line_without_colon_skipped(self):
+        self.assertEqual(
+            api_client._parse_content_length(
+                "HTTP/1.1 200 OK\r\nGARBAGE-LINE\r\nContent-Length: 2"),
+            2)
+
+    def test_unix_api_with_payload_and_chunked_body(self):
+        # 体部分两片到达：首片随头一起到（体部前缀），次片
+        # 才补满 Content-Length——_read_body 的追加行必须跑到。
+        fake = _FakeApiSocket([
+            b"HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nab",
+            b"cd",
+        ])
+        with tempfile.TemporaryDirectory() as tmp:
+            sock_path = os.path.join(tmp, "chv.sock")
+            with open(sock_path, "w") as fh:
+                fh.write("x")  # 仅让 exists() 为真，不真建 socket
+            with mock.patch.object(api_client.socket, "socket",
+                                   lambda *a, **k: fake):
+                status, body = api_client._unix_socket_api(
+                    sock_path, "PUT", "/vm.create",
+                    payload={"name": "vm1"})
+        self.assertEqual(status, 200)
+        self.assertEqual(body, "abcd")
+        self.assertIn(b'"name": "vm1"', fake.sent)
+        self.assertTrue(fake.closed)
+
+    def test_parse_status_non_numeric_is_zero(self):
+        status, _body = api_client._parse_response(
+            "HTTP/1.1 OK\r\n\r\n", b"{}")
+        self.assertEqual(status, 0)
+
+    def test_parse_status_missing_token_is_zero(self):
+        status, _body = api_client._parse_response("HTTP/\r\n\r\n", b"{}")
+        self.assertEqual(status, 0)
+
+    def test_normalize_dict_branches(self):
+        self.assertEqual(
+            api_client.normalize_api_result(
+                {"status": 201, "body": {"a": 1}}),
+            (201, {"a": 1}))
+        self.assertEqual(
+            api_client.normalize_api_result({"status_code": 202}),
+            (202, {"status_code": 202}))
+        self.assertEqual(
+            api_client.normalize_api_result({"body": "x"}),
+            (0, "x"))
+
+    def test_normalize_object_branches(self):
+        obj = SimpleNamespace(status=404)
+        status, body = api_client.normalize_api_result(obj)
+        self.assertEqual((status, body), (404, obj))
+        obj2 = SimpleNamespace(status_code=503)
+        status, _body = api_client.normalize_api_result(obj2)
+        self.assertEqual(status, 503)
+        obj3 = SimpleNamespace()
+        status, _body = api_client.normalize_api_result(obj3)
+        self.assertEqual(status, 0)
+
+
+class Compat215Cover100Test(unittest.TestCase):
+    def test_commonpath_value_error_is_none(self):
+        # 跨盘符（Windows）下 commonpath 抛 ValueError 时
+        # 目录门必须回 None，而不是把异常漏给调用方。
+        with tempfile.TemporaryDirectory() as tmp:
+            with mock.patch.object(
+                    os.path, "commonpath",
+                    side_effect=ValueError("mixing drives")):
+                self.assertIsNone(compat215.resolve_inside(tmp, "sub"))
+
+
+class ConfigCover100Test(unittest.TestCase):
+    def test_disk_gb_non_numeric_rejected(self):
+        with self.assertRaises(ValueError) as ctx:
+            chv_config.build_chv_config({}, {"CHV_DISK_GB": "abc"})
+        self.assertIn("CHV_DISK_GB 必须是数字", str(ctx.exception))
+
+    def test_snapshot_count_non_numeric_rejected(self):
+        with self.assertRaises(ValueError) as ctx:
+            chv_config.build_chv_config(
+                {}, {"CHV_SNAPSHOT_MAX_COUNT": "x"})
+        self.assertIn("CHV_SNAPSHOT_* 必须是数字", str(ctx.exception))
+
+    def test_ready_interval_zero_falls_back_to_five(self):
+        cfg = chv_config.build_chv_config({"ready_interval": 0}, {})
+        self.assertEqual(cfg.ready_interval, 5)
+
+
+class DiskCover100Test(unittest.TestCase):
+    def _write(self, path, data):
+        os.makedirs(os.path.dirname(path), exist_ok=True)
+        with open(path, "wb") as fh:
+            fh.write(data)
+
+    def test_dir_total_ignores_vanished_file(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rootfs = os.path.join(tmp, "base.raw")
+            self._write(rootfs, b"1234")
+            state_dir = os.path.join(tmp, "state")
+            victim = os.path.join(state_dir, "gone.raw")
+            self._write(victim, b"zz")
+            real_getsize = os.path.getsize
+
+            def getsize(path):
+                if path == victim:
+                    raise OSError("walk 后文件消失")
+                return real_getsize(path)
+
+            with mock.patch.object(os.path, "getsize",
+                                   side_effect=getsize):
+                # 不抛错：消失文件按 0 计，不卡住配额校验。
+                self.assertIsNone(
+                    chv_disk.check_disk_quota(rootfs, 50.0, state_dir))
+
+    def test_quota_skipped_without_disk_gb_or_rootfs(self):
+        self.assertIsNone(chv_disk.check_disk_quota("", None))
+        self.assertIsNone(chv_disk.check_disk_quota("anything", None))
+        self.assertIsNone(chv_disk.check_disk_quota("", 50.0))
+
+    def test_quota_disk_usage_failure_is_not_fatal(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rootfs = os.path.join(tmp, "base.raw")
+            self._write(rootfs, b"1234")
+            with mock.patch.object(chv_disk.shutil, "disk_usage",
+                                   side_effect=OSError("statfs 失败")):
+                # 宿主余量读不到时不按“不足”误杀，只按已用量
+                # 与镜像大小校验，本例两者都远小于配额。
+                self.assertIsNone(
+                    chv_disk.check_disk_quota(rootfs, 50.0, tmp))
+
+    def test_quota_host_free_space_shortfall_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rootfs = os.path.join(tmp, "base.raw")
+            self._write(rootfs, b"1234")
+            usage = SimpleNamespace(total=100, used=100, free=0)
+            with mock.patch.object(chv_disk.shutil, "disk_usage",
+                                   return_value=usage):
+                with self.assertRaises(ChvFailure) as ctx:
+                    chv_disk.check_disk_quota(rootfs, 50.0, tmp)
+        self.assertEqual(ctx.exception.kind, "disk_quota_exceeded")
+
+    def test_provision_existing_disk_unreadable_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rootfs = os.path.join(tmp, "base.raw")
+            self._write(rootfs, b"1234")
+            task_disk = os.path.join(tmp, "vm1", "disk.raw")
+            self._write(task_disk, b"1234")
+            with mock.patch.object(os.path, "getsize",
+                                   side_effect=OSError("不可读")):
+                with self.assertRaises(ChvFailure) as ctx:
+                    chv_disk.provision_task_disk(tmp, "vm1", rootfs,
+                                                 "copy")
+        self.assertEqual(ctx.exception.kind, "disk_quota_unverified")
+
+    def test_provision_copy_failure_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            rootfs = os.path.join(tmp, "base.raw")
+            self._write(rootfs, b"1234")
+            with mock.patch.object(chv_disk.shutil, "copyfile",
+                                   side_effect=OSError("磁盘满")):
+                with self.assertRaises(ChvFailure) as ctx:
+                    chv_disk.provision_task_disk(tmp, "vm1", rootfs,
+                                                 "copy")
+        self.assertEqual(ctx.exception.kind, "disk_quota_unverified")
+
+    def test_prune_delete_failure_is_best_effort(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            snap = os.path.join(tmp, "vm1", "old.snap")
+            self._write(snap, b"old")
+            chv_disk._save_snapshots(tmp, "vm1", [{
+                "url": "file://" + snap, "path": snap, "size": 3,
+                "created_at": 0}])
+            with mock.patch.object(os, "remove",
+                                   side_effect=OSError("删不掉")):
+                kept, pruned = chv_disk.prune_snapshots(
+                    tmp, "vm1", retention_days=30, now=10 ** 9)
+            self.assertEqual(kept, [])
+            self.assertEqual(len(pruned), 1)
+            self.assertTrue(os.path.exists(snap))
+
+    def test_snapshot_limits_uses_file_size_when_size_missing(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            snap = os.path.join(tmp, "vm1", "a.snap")
+            self._write(snap, b"12345")
+            chv_disk._save_snapshots(tmp, "vm1", [{
+                "url": "file://" + snap, "path": snap,
+                "created_at": 0}])
+            kept = chv_disk.check_snapshot_limits(
+                tmp, "vm1", max_count=10, max_total_gb=50.0,
+                retention_days=None)
+            self.assertEqual(len(kept), 1)
+
+    def test_snapshot_limits_ignores_unreadable_file_size(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            snap = os.path.join(tmp, "vm1", "a.snap")
+            self._write(snap, b"12345")
+            chv_disk._save_snapshots(tmp, "vm1", [{
+                "url": "file://" + snap, "path": snap,
+                "created_at": 0}])
+            with mock.patch.object(os.path, "getsize",
+                                   side_effect=OSError("不可读")):
+                kept = chv_disk.check_snapshot_limits(
+                    tmp, "vm1", max_count=10, max_total_gb=50.0,
+                    retention_days=None)
+            self.assertEqual(len(kept), 1)
+
+    def test_snapshot_limits_total_cap_via_file_size_rejected(self):
+        # 总量统计走文件大小回落时同样触发总量帽：无 size
+        # 字段的条目按实际文件大小计入，达到上限即拒绝。
+        with tempfile.TemporaryDirectory() as tmp:
+            snap = os.path.join(tmp, "vm1", "a.snap")
+            self._write(snap, b"12345")
+            chv_disk._save_snapshots(tmp, "vm1", [{
+                "url": "file://" + snap, "path": snap,
+                "created_at": 0}])
+            with self.assertRaises(ChvFailure) as ctx:
+                chv_disk.check_snapshot_limits(
+                    tmp, "vm1", max_count=10, max_total_gb=0.0,
+                    retention_days=None)
+        self.assertEqual(ctx.exception.kind, "snapshot_quota_exceeded")
+
+    def test_record_file_snapshot_missing_file_size_none(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            entries, _removed = chv_disk.record_snapshot(
+                tmp, "vm1", "file:///nonexistent-chv-snap", 50.0)
+            self.assertEqual(len(entries), 1)
+            self.assertIsNone(entries[0]["size"])
+            self.assertEqual(entries[0]["path"], "/nonexistent-chv-snap")
+
+    def test_record_cap_removal_delete_failure_is_best_effort(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            old = os.path.join(tmp, "vm1", "old.snap")
+            self._write(old, b"old")
+            chv_disk._save_snapshots(tmp, "vm1", [{
+                "url": "file://" + old, "path": old, "size": 10 ** 6,
+                "created_at": 0}])
+            real_remove = os.remove
+
+            def remove(path):
+                if path == old:
+                    raise OSError("删不掉")
+                return real_remove(path)
+
+            with mock.patch.object(os, "remove", side_effect=remove):
+                entries, removed = chv_disk.record_snapshot(
+                    tmp, "vm1", "https://example.invalid/new", 0.0)
+            self.assertEqual(len(removed), 1)
+            self.assertEqual(removed[0]["path"], old)
+            self.assertEqual(len(entries), 1)
+            self.assertTrue(os.path.exists(old))
+
+
+class HardeningCover100Test(unittest.TestCase):
+    def test_state_dir_stat_failure_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            state_dir = os.path.join(tmp, "state")
+            real_stat = os.stat
+
+            def stat(path, *args, **kwargs):
+                if path == state_dir:
+                    raise OSError("stat 失败")
+                return real_stat(path, *args, **kwargs)
+
+            with mock.patch.object(os, "stat", side_effect=stat):
+                with self.assertRaises(ChvFailure) as ctx:
+                    hardening.ensure_secure_state_dir(state_dir)
+        self.assertEqual(ctx.exception.kind, "state_dir_insecure")
+
+    def test_state_dir_still_loose_after_chmod_rejected(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            state_dir = os.path.join(tmp, "state")
+            os.makedirs(state_dir)
+            os.chmod(state_dir, 0o755)
+            with mock.patch.object(os, "chmod",
+                                   lambda *a, **k: None):
+                with self.assertRaises(ChvFailure) as ctx:
+                    hardening.ensure_secure_state_dir(state_dir)
+        self.assertEqual(ctx.exception.kind, "state_dir_insecure")
+
+    def test_socket_is_live_empty_and_missing_is_dead(self):
+        self.assertFalse(hardening.socket_is_live(""))
+        self.assertFalse(hardening.socket_is_live(None))
+        self.assertFalse(
+            hardening.socket_is_live("/tmp/no-such-chv.sock"))
+
+    def test_socket_is_live_socket_error_is_dead(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "dead.sock")
+            with open(path, "w") as fh:
+                fh.write("x")
+            with mock.patch.object(hardening.socket, "socket",
+                                   side_effect=OSError("socket 失败")):
+                self.assertFalse(hardening.socket_is_live(path))
+
+    def test_parse_net_entries_empty_is_empty(self):
+        self.assertEqual(hardening.parse_net_entries(""), [])
+        self.assertEqual(hardening.parse_net_entries(None), [])
+
+    def test_parse_net_entries_non_dict_rejected(self):
+        with self.assertRaises(ChvFailure) as ctx:
+            hardening.parse_net_entries('[1, 2]')
+        self.assertEqual(ctx.exception.kind, "net_config_invalid")
+
+    def test_egress_proxy_empty_is_none(self):
+        self.assertIsNone(hardening.validate_egress_proxy(""))
+        self.assertIsNone(hardening.validate_egress_proxy(None))
+
+    def test_egress_proxy_unparsable_rejected(self):
+        with mock.patch("urllib.parse.urlparse",
+                        side_effect=ValueError("bad url")):
+            with self.assertRaises(ChvFailure) as ctx:
+                hardening.validate_egress_proxy("http://[::1")
+        self.assertEqual(ctx.exception.kind, "config_invalid")
+
+
+class HostkeysCover100Test(unittest.TestCase):
+    def test_build_seed_disk_subprocess_failure(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "seed.iso")
+            result = SimpleNamespace(returncode=3, stderr="boom")
+            with mock.patch("subprocess.run", return_value=result):
+                with self.assertRaises(RuntimeError) as ctx:
+                    hostkeys.build_seed_disk(tmp, out, runner=None)
+        self.assertIn("genisoimage 失败", str(ctx.exception))
+
+    def test_build_seed_disk_chmod_failure(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            out = os.path.join(tmp, "seed.iso")
+
+            def runner(cmd):
+                with open(out, "wb") as fh:
+                    fh.write(b"ISO")
+                return SimpleNamespace(returncode=0)
+
+            with mock.patch.object(os, "chmod",
+                                   side_effect=OSError("chmod 失败")):
+                with self.assertRaises(RuntimeError) as ctx:
+                    hostkeys.build_seed_disk(tmp, out, runner=runner)
+        self.assertIn("chmod 0600 失败", str(ctx.exception))
+
+    def test_write_seed_without_runcmd_still_appends_secret(self):
+        # 防御分支：若 user-data 模板将来没有 runcmd 段，
+        # 密钥环境条目必须追加而不是丢弃。
+        with tempfile.TemporaryDirectory() as tmp:
+            seed_dir = os.path.join(tmp, "seed")
+            with mock.patch.object(
+                    hostkeys, "build_seed_user_data",
+                    return_value="#cloud-config\nwrite_files:\n"):
+                hostkeys.write_seed(seed_dir, "vm1", "PRIV", "PUB",
+                                    secret_env={"MY_KEY": "a b"})
+            with open(os.path.join(seed_dir, "user-data")) as fh:
+                content = fh.read()
+        self.assertIn(hostkeys.GUEST_ENV_PATH, content)
+        self.assertIn("MY_KEY=", content)
+
+    def test_keygen_subprocess_failure(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            result = SimpleNamespace(returncode=1, stderr="boom")
+            with mock.patch("subprocess.run", return_value=result):
+                with self.assertRaises(RuntimeError) as ctx:
+                    hostkeys.generate_host_keypair(tmp, "vm1",
+                                                   runner=None)
+        self.assertIn("ssh-keygen 失败", str(ctx.exception))
+
+    def test_keygen_private_chmod_failure(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            paths = hostkeys.host_key_paths(tmp, "vm1")
+
+            def runner(cmd):
+                os.makedirs(os.path.dirname(paths["private"]),
+                            exist_ok=True)
+                for key in ("private", "public"):
+                    with open(paths[key], "w") as fh:
+                        fh.write("x")
+                return SimpleNamespace(returncode=0)
+
+            with mock.patch.object(os, "chmod",
+                                   side_effect=OSError("chmod 失败")):
+                with self.assertRaises(RuntimeError) as ctx:
+                    hostkeys.generate_host_keypair(tmp, "vm1",
+                                                   runner=runner)
+        self.assertIn("chmod 0600 失败", str(ctx.exception))
+
+
+class _PoolStub:
+    """WarmPool 用的假执行器：行为由属性开关控制。"""
+
+    def __init__(self, registry_path=None):
+        self.state_dir = None
+        self.warm_pool_size = 0
+        self.snapshot_url = "file:///snap/base"
+        self.registry_path = registry_path
+        self.shutdown_calls = []
+        self.shutdown_raises = False
+        self.emit_raises = False
+        self.woken = []
+        self.wake_raises = False
+
+    def vm_name(self, task):
+        raw = task.get("id") if isinstance(task, dict) else str(task)
+        return raw if raw.startswith("chv-") else "chv-" + raw
+
+    def wake_vm(self, task):
+        if self.wake_raises:
+            raise RuntimeError("wake 失败")
+        self.woken.append(task["id"])
+        return {"vm": self.vm_name(task)}
+
+    def pause_vm(self, task):
+        return True
+
+    def resume_vm(self, task):
+        return True
+
+    def shutdown_vm(self, task_or_name):
+        self.shutdown_calls.append(task_or_name)
+        if self.shutdown_raises:
+            raise RuntimeError("shutdown 失败")
+        return True
+
+    def _emit(self, event, task, detail="", kind=None):
+        if self.emit_raises:
+            raise RuntimeError("audit 失败")
+
+
+class ExperimentalPoolCover100Test(unittest.TestCase):
+    """实验池覆盖（单独计分，不计入生产覆盖）。"""
+
+    def test_load_registry_invalid_entry_counted(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "warm-pool.json")
+            with open(path, "w") as fh:
+                json.dump({
+                    "bad": "not-a-dict",
+                    "good": {"vm": "vm-good", "state": "warm",
+                             "task_id": "t1",
+                             "snapshot_url": None},
+                }, fh)
+            pool = WarmPool(_PoolStub(), registry_path=path)
+        self.assertEqual(pool.registry_corrupt_entries, 1)
+        self.assertEqual(pool.orphan_count(), 1)
+        self.assertEqual(pool.warm_count(), 0)
+
+    def test_save_registry_skips_non_warm_entries(self):
+        # orphan 是上一世待清理条目，落盘时不写回：重启后
+        # 孤儿改由 socket 扫描兜底，不借 registry 复活。
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "warm-pool.json")
+            pool = WarmPool(_PoolStub(), registry_path=path)
+            pool._entries["w"] = {"state": "warm", "vm": "w",
+                                  "task": {"id": "w"},
+                                  "snapshot_url": None}
+            pool._entries["old"] = {"state": "orphan", "vm": "old",
+                                    "task": {"id": "old"},
+                                    "snapshot_url": None}
+            pool._save_registry()
+            with open(path) as fh:
+                data = json.load(fh)
+        self.assertEqual(list(data), ["w"])
+
+    def test_save_registry_non_dict_task_id_none(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "warm-pool.json")
+            pool = WarmPool(_PoolStub(), registry_path=path)
+            pool._entries["w"] = {"state": "warm", "vm": "w",
+                                  "task": "not-a-dict",
+                                  "snapshot_url": None}
+            pool._save_registry()
+            with open(path) as fh:
+                data = json.load(fh)
+        self.assertIsNone(data["w"]["task_id"])
+
+    def test_reconcile_skips_warm_entries(self):
+        pool = WarmPool(_PoolStub(), registry_path=None)
+        pool._entries["warm-1"] = {"state": "warm", "vm": "vm-warm",
+                                   "task": {"id": "warm-1"}}
+        pool._entries["old-1"] = {"state": "orphan", "vm": "vm-old",
+                                  "task": {"id": "old-1"}}
+        cleaned = pool.reconcile_orphans()
+        self.assertEqual(cleaned, ["vm-old"])
+        self.assertIn("warm-1", pool._entries)
+        self.assertNotIn("old-1", pool._entries)
+
+    def test_reconcile_shutdown_failure_still_cleans(self):
+        stub = _PoolStub()
+        stub.shutdown_raises = True
+        pool = WarmPool(stub, registry_path=None)
+        pool._entries["old-1"] = {"state": "orphan", "vm": "vm-old",
+                                  "task": {"id": "old-1"}}
+        self.assertEqual(pool.reconcile_orphans(), ["vm-old"])
+        self.assertEqual(pool._entries, {})
+
+    def test_reconcile_emit_failure_is_best_effort(self):
+        stub = _PoolStub()
+        stub.emit_raises = True
+        pool = WarmPool(stub, registry_path=None)
+        pool._entries["old-1"] = {"state": "orphan", "vm": "vm-old",
+                                  "task": {"id": "old-1"}}
+        self.assertEqual(pool.reconcile_orphans(), ["vm-old"])
+
+    def test_prewarm_skips_colliding_name(self):
+        stub = _PoolStub()
+        pool = WarmPool(stub, size=2, registry_path=None)
+        pool._entries["warm-1"] = {"state": "warm", "vm": "vm-1",
+                                   "task": {"id": "warm-1"},
+                                   "snapshot_url": pool.snapshot_url}
+        added = pool.prewarm()
+        self.assertEqual(added, ["warm-2"])
+        self.assertEqual(stub.woken, ["warm-2"])
+        self.assertEqual(pool.warm_count(), 2)
+
+    def test_prewarm_wake_failure_drops_entry_and_raises(self):
+        stub = _PoolStub()
+        stub.wake_raises = True
+        pool = WarmPool(stub, size=1, registry_path=None)
+        with self.assertRaises(RuntimeError):
+            pool.prewarm()
+        self.assertEqual(pool._entries, {})
+        self.assertEqual(pool.warm_count(), 0)
+
+    def test_acquire_emit_failure_is_best_effort(self):
+        stub = _PoolStub()
+        stub.emit_raises = True
+        pool = WarmPool(stub, size=1, registry_path=None)
+        pool.prewarm()
+        got = pool.acquire({"id": "t1"})
+        self.assertEqual(got["mode"], "warm")
+
+    def test_release_shutdown_failure_still_reports_destroyed(self):
+        stub = _PoolStub()
+        stub.shutdown_raises = True
+        pool = WarmPool(stub, size=1, registry_path=None)
+        pool._entries["warm-1"] = {"state": "assigned", "vm": "vm-1",
+                                   "task": {"id": "warm-1"}}
+        self.assertTrue(pool.release({"id": "t1"}, pool_vm="warm-1"))
+        self.assertNotIn("warm-1", pool._entries)
+
+
+if __name__ == "__main__":
+    unittest.main()