"""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()