622 lines
26 KiB
Python
622 lines
26 KiB
Python
"""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_not_cleaned(self):
|
||
# feat-chv-p1fix（P1-01 同款语义，实验池）：shutdown
|
||
# 失败的条目不进 cleaned、不从内存条目里 pop，
|
||
# 下次 reconcile 可重试。
|
||
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(), [])
|
||
self.assertIn("old-1", 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()
|