Files
codex-app-server/cloud/tests/test_chv_arch_maint_fe.py
T
host 78098c5594 feat(chv-muse-p1): 常驻 VM + WarmPool 扶正 + 气球 + vsock,对齐 Muse
每用户一台常驻 VM(暂停态,resume 到 SSH 5 秒内为待验证目标),
池空/失败回落冷启动;CHV 配置加 virtio-balloon + free-page-reporting
与 vsock notify 最小实现(SSH 仍是命令通道与就绪判定)。
新模块 resident_pool.py / vsock.py + 客内 vsock_notify.py;
_config 增加 CHV_RESIDENT_IDLE_MINUTES / CHV_BALLOON / CHV_VSOCK;
worker_adapter 接常驻孤儿清理 facade;ADR-17 差距表见
docs/chv-muse-p1-gap.md。

恢复收尾(分离后回归,见 gap.md §4):VmDeathTest 死亡模拟改杀
实际跑命令的 VM;两处冷路径旧测试显式关常驻;_balloon_vsock_payload
与 listen_notify 拆到 50 行内;resident_pool.py:208 mypy 修掉。
executor.py out 未注解与 marker_tools.py:142 是基线旧错,未动。

测试:全量 cloud 套件 1934 OK(skipped=8);mypy 在改动文件上
干净(仅基线旧错 4 处)。真机验证待第三棒让出 WSL 测试机。
2026-11-06 21:31:48 +00:00

471 lines
20 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""CHV 架构/可维护性/前端(只读侧)推进轨测试。
本轨 feat-chv-arch-maint-fe,堆叠于 #254(feat-chv-
arch95plus)。只用标准库与 mock,绝不启动真实 VM;
测试里只用 FAKE 占位值,绝无真实密钥。覆盖:
- WarmPool 破环性清理:旧 import 路径一律 ImportError,
实验路径唯一,生产代码不 import 实验模块;
- orphans facade:registry 语义(有效/非法/损坏/缺失/
shutdown 抛错/_emit 抛错)与旧 WarmPool 逐项一致;
- 配置片段:五片段键集互不相交、并集等于旧字段集、
默认值与单项非法报错逐字一致、组装键冲突即报错;
- ADR-12 模块依赖图:叶子模块只依赖标准库、禁 import
清单、公开 API 全量类型标注且无 Any 逃逸;
- #215 替换清单:每项 local 真实存在、#215 模块当前
不存在、R3 永不替换;
- 服务端只读字段:goal 回落形态(纯文本“VM 中途死亡”)
同样派生 chv_vm_dead,非 VM 中断原因一律不派生。
"""
import ast
import importlib.util
import json
import os
import sys
import tempfile
import unittest
from pathlib import Path
try:
from cloud.chv import reconcile_orphans_for_executor
from cloud.chv.compat215 import REPLACEMENT_POINTS
from cloud.chv.compat215 import resolve_inside, tmpfs_isolated
from cloud.chv.config import _assemble_chv_config
from cloud.chv.config import _balloon_vsock_fragment
from cloud.chv.config import _base_fragment
from cloud.chv.config import _disk_fragment
from cloud.chv.config import _network_fragment
from cloud.chv.config import _quota_fragment
from cloud.chv.config import _ssh_fragment
from cloud.chv.config import _timeout_fragment
from cloud.chv.config import build_chv_config
from cloud.chv.experimental_pool import WarmPool
from cloud.chv.orphans import _REGISTRY_FILENAME
from cloud.chv.orphans import reconcile_orphans_for_executor as _facade
from cloud.server import _chv_vm_dead_from_row, _task_to_dict
except ImportError:
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from cloud.chv import reconcile_orphans_for_executor
from cloud.chv.compat215 import REPLACEMENT_POINTS
from cloud.chv.compat215 import resolve_inside, tmpfs_isolated
from cloud.chv.config import _assemble_chv_config
from cloud.chv.config import _balloon_vsock_fragment
from cloud.chv.config import _base_fragment
from cloud.chv.config import _disk_fragment
from cloud.chv.config import _network_fragment
from cloud.chv.config import _quota_fragment
from cloud.chv.config import _ssh_fragment
from cloud.chv.config import _timeout_fragment
from cloud.chv.config import build_chv_config
from cloud.chv.experimental_pool import WarmPool
from cloud.chv.orphans import _REGISTRY_FILENAME
from cloud.chv.orphans import reconcile_orphans_for_executor as _facade
from cloud.server import _chv_vm_dead_from_row, _task_to_dict
ROOT = Path(__file__).resolve().parents[2]
CHV = ROOT / "cloud" / "chv"
EXPECTED_FIELDS = {
"binary", "kernel", "rootfs", "cpus", "memory", "state_dir",
"kvm_path", "cpuinfo_path", "ch_remote", "model",
"ssh_key", "ssh_user", "ssh_port", "host_key_mode",
"guest_host_py", "cmdline",
"net", "net_json", "ssh_host", "ssh_forward",
"egress_proxy", "allow_direct_egress",
"max_vms", "warm_pool_size",
"disk_mode", "disk_gb", "reuse_experiment",
"single_tenant_project", "snapshot_max_count",
"snapshot_max_total_gb", "snapshot_retention_days",
"snapshot_dir", "allow_remote_snapshot", "snapshot_url",
"task_timeout", "api_timeout", "ready_timeout", "ready_interval",
# P1-B(feat-chv-muse-p1):气球/vsock 三字段。
"balloon", "vsock", "vsock_cid",
}
class FakeProdExecutor:
"""生产 facade 替身:只有 facade 允许碰的属性。
acquire/prewarm/release 一律不存在——facade 若敢碰,
__getattr__ 直接抛错,测试即 fail。"""
def __init__(self, state_dir):
self.state_dir = state_dir
self.shutdowns = []
self.emits = []
def __getattr__(self, name):
if name in ("acquire", "prewarm", "release"):
raise AssertionError("facade 不得碰实验接口:%s" % name)
raise AttributeError(name)
def shutdown_vm(self, vm):
self.shutdowns.append(vm)
return True
def _emit(self, event, task, detail):
self.emits.append(event)
def _write_registry(state_dir, data):
path = os.path.join(state_dir, _REGISTRY_FILENAME)
if isinstance(data, str):
with open(path, "w") as fh:
fh.write(data)
else:
with open(path, "w") as fh:
json.dump(data, fh)
return path
class BreakingCleanupTest(unittest.TestCase):
def test_old_paths_raise_import_error(self):
with self.assertRaises(ImportError):
from cloud.chv import WarmPool # noqa: F401
with self.assertRaises(ImportError):
from cloud.sandbox import WarmPool # noqa: F401
with self.assertRaises(ImportError):
import cloud.chv.pool # noqa: F401
def test_new_paths_and_identity(self):
self.assertEqual(WarmPool.__module__,
"cloud.chv.experimental_pool")
self.assertIs(reconcile_orphans_for_executor, _facade)
import cloud.sandbox as sb
self.assertIs(sb.reconcile_orphans_for_executor, _facade)
self.assertFalse(WarmPool.wired_to_dispatch)
def test_production_sources_never_import_experimental(self):
# 钉的是 import 语句,不是文档提及:executor/__init__
# 的 docstring 本来就要写清实验模块去向,文本匹配会
# 误杀,故用 AST 只看 Import/ImportFrom 节点。
for rel in ("cloud/worker.py", "cloud/chv/executor.py",
"cloud/chv/__init__.py", "cloud/sandbox.py",
"cloud/chv/orphans.py"):
tree = ast.parse((ROOT / rel).read_text())
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for a in node.names:
self.assertNotIn(a.name, (
"cloud.chv.pool",
"cloud.chv.experimental_pool"), rel)
if isinstance(node, ast.ImportFrom):
self.assertNotIn(node.module, (
"cloud.chv.pool",
"cloud.chv.experimental_pool"), rel)
if node.module in ("cloud.chv", "cloud.sandbox"):
names = [a.name for a in node.names]
self.assertNotIn("WarmPool", names, rel)
class OrphansTest(unittest.TestCase):
def test_valid_warm_and_assigned_cleaned_invalid_dropped(self):
with tempfile.TemporaryDirectory() as d:
path = _write_registry(d, {
"a": {"state": "warm", "vm": "chv-a",
"task_id": "a", "snapshot_url": None},
"b": {"state": "assigned", "vm": "chv-b",
"task_id": "b", "snapshot_url": None},
"c": {"state": "warm", "task_id": "c"},
"e": {"state": "running", "vm": "chv-e"},
"f": "not-a-dict",
})
ex = FakeProdExecutor(d)
self.assertEqual(
reconcile_orphans_for_executor(ex),
["chv-a", "chv-b"])
self.assertEqual(ex.shutdowns, ["chv-a", "chv-b"])
self.assertEqual(ex.emits, ["warm_pool_reconcile"])
with open(path) as fh:
self.assertEqual(json.load(fh), {})
def test_invalid_only_returns_empty_and_leaves_file(self):
with tempfile.TemporaryDirectory() as d:
path = _write_registry(d, {
"c": {"state": "warm", "task_id": "c"}})
ex = FakeProdExecutor(d)
self.assertEqual(reconcile_orphans_for_executor(ex), [])
self.assertEqual(ex.shutdowns, [])
with open(path) as fh:
self.assertIn("task_id", fh.read())
def test_corrupt_json_quarantined(self):
with tempfile.TemporaryDirectory() as d:
path = _write_registry(d, "{not json")
ex = FakeProdExecutor(d)
self.assertEqual(reconcile_orphans_for_executor(ex), [])
self.assertFalse(os.path.exists(path))
self.assertTrue(os.path.exists(path + ".corrupt"))
def test_nondict_top_level_quarantined(self):
with tempfile.TemporaryDirectory() as d:
path = _write_registry(d, "[1, 2]")
ex = FakeProdExecutor(d)
self.assertEqual(reconcile_orphans_for_executor(ex), [])
self.assertTrue(os.path.exists(path + ".corrupt"))
def test_missing_file_and_no_state_dir(self):
with tempfile.TemporaryDirectory() as d:
ex = FakeProdExecutor(d)
self.assertEqual(reconcile_orphans_for_executor(ex), [])
class NoDir:
def shutdown_vm(self, vm):
raise AssertionError("must not be called")
self.assertEqual(
reconcile_orphans_for_executor(NoDir()), [])
def test_shutdown_and_emit_failures_do_not_break_cleanup(self):
with tempfile.TemporaryDirectory() as d:
_write_registry(d, {
"a": {"state": "warm", "vm": "chv-a"}})
ex = FakeProdExecutor(d)
def boom(vm):
raise RuntimeError("shutdown down")
ex.shutdown_vm = boom
ex._emit = lambda *a, **k: (_ for _ in ()).throw(
RuntimeError("audit down"))
self.assertEqual(
reconcile_orphans_for_executor(ex), ["chv-a"])
def test_experimental_pool_registry_roundtrip_unchanged(self):
# 实验池自身的 registry 语义(载入即 orphan、reconcile
# 清理)与上一轨一致,只是文件 IO 委托给 orphans。
with tempfile.TemporaryDirectory() as d:
ex = FakeProdExecutor(d)
ex.warm_pool_size = 0
ex.snapshot_url = None
pool1 = WarmPool(ex)
pool1._entries["chv-old"] = {
"state": "warm", "vm": "chv-old",
"task": {"id": "old"}, "snapshot_url": None}
pool1._save_registry()
pool2 = WarmPool(ex)
self.assertEqual(pool2.orphan_count(), 1)
self.assertEqual(pool2.warm_count(), 0)
self.assertEqual(pool2.reconcile_orphans(), ["chv-old"])
class ConfigFragmentsTest(unittest.TestCase):
def test_fragment_keys_disjoint_and_union_is_field_set(self):
frags = [_base_fragment({}, {}), _ssh_fragment({}, {}),
_network_fragment({}, {}), _quota_fragment({}),
_disk_fragment({}), _timeout_fragment({}, {}),
# P1-B:新片段加入并集断言(字段 38 → 41)。
_balloon_vsock_fragment({})]
seen = set()
for frag in frags:
self.assertFalse(seen & set(frag), "片段键重叠")
seen |= set(frag)
self.assertEqual(seen, EXPECTED_FIELDS)
self.assertEqual(len(EXPECTED_FIELDS), 41)
def test_defaults_and_overrides(self):
cfg = build_chv_config({}, {})
self.assertEqual(cfg.binary, "cloud-hypervisor")
self.assertEqual(cfg.ssh_port, 2222)
self.assertEqual(cfg.ssh_user, "root")
self.assertEqual(cfg.host_key_mode, "provisioned")
self.assertEqual(cfg.max_vms, 4)
self.assertEqual(cfg.disk_mode, "copy")
self.assertEqual(cfg.disk_gb, 50.0)
self.assertEqual(cfg.api_timeout, 30.0)
self.assertEqual(cfg.ready_timeout, 120.0)
self.assertEqual(cfg.ready_interval, 5.0)
self.assertIsNone(cfg.task_timeout)
self.assertIsNone(cfg.snapshot_url)
cfg2 = build_chv_config(
{"ssh_user": "u", "ssh_port": 2223},
{"CHV_SSH_USER": "env-user", "CHV_MAX_VMS": "2",
"CHV_EGRESS_PROXY": "http://p:1",
"CHV_SNAPSHOT_URL": "file:///snap/base"})
self.assertEqual(cfg2.ssh_user, "u")
self.assertEqual(cfg2.ssh_port, 2223)
self.assertEqual(cfg2.max_vms, 2)
self.assertEqual(cfg2.egress_proxy, "http://p:1")
self.assertEqual(cfg2.snapshot_url, "file:///snap/base")
def test_single_error_messages_unchanged(self):
cases = [
({"ssh_user": ""}, {}, "CHV_SSH_USER 不能为空字符串"),
({}, {"CHV_SSH_PORT": "0"}, "CHV_SSH_PORT 必须在 1-65535 之间"),
({}, {"CHV_SSH_PORT": "99999"},
"CHV_SSH_PORT 必须在 1-65535 之间"),
({}, {"CHV_DISK_MODE": "bad"},
"CHV_DISK_MODE 只能是 copy 或 reuse"),
({}, {"CHV_DISK_GB": "0"}, "CHV_DISK_GB 必须大于 0"),
({}, {"CHV_SNAPSHOT_MAX_COUNT": "-1"},
"CHV_SNAPSHOT_* 不能为负数"),
({}, {"CHV_MAX_VMS": "x"}, "CHV_MAX_VMS 必须是整数"),
({}, {"CHV_WARM_POOL_SIZE": "x"},
"CHV_WARM_POOL_SIZE 必须是整数"),
({}, {"CHV_API_TIMEOUT": "x"},
"CHV_*_TIMEOUT/INTERVAL 必须是数字"),
]
for params, env, msg in cases:
with self.assertRaises(ValueError) as cm:
build_chv_config(params, env)
self.assertIn(msg, str(cm.exception))
def test_assemble_key_conflict_raises(self):
with self.assertRaises(ValueError):
_assemble_chv_config([{"a": 1}, {"a": 2}])
def _top_level_imports(path):
tree = ast.parse(path.read_text())
mods = set()
for node in tree.body:
if isinstance(node, ast.Import):
mods.update(a.name for a in node.names)
elif isinstance(node, ast.ImportFrom) and node.module:
mods.add(node.module)
return mods
class DependencyGraphTest(unittest.TestCase):
def test_leaf_modules_only_import_stdlib(self):
for name in ("orphans", "config", "compat215", "errors"):
mods = _top_level_imports(CHV / (name + ".py"))
for mod in mods:
top = mod.split(".")[0]
self.assertFalse(top.startswith("cloud"),
"%s 不得 import %s" % (name, mod))
def test_executor_never_imports_pool_or_orphans(self):
# 同上:只看 AST import 节点,模块头 docstring 的
# 分工说明不算 import。
tree = ast.parse((CHV / "executor.py").read_text())
for node in ast.walk(tree):
if isinstance(node, ast.Import):
for a in node.names:
self.assertNotIn(a.name, (
"cloud.chv.pool",
"cloud.chv.experimental_pool",
"cloud.chv.orphans"))
if isinstance(node, ast.ImportFrom):
self.assertNotIn(node.module, (
"cloud.chv.pool",
"cloud.chv.experimental_pool",
"cloud.chv.orphans"))
def test_no_chv_function_over_50_lines(self):
over = []
for f in CHV.glob("*.py"):
tree = ast.parse(f.read_text())
for node in ast.walk(tree):
if isinstance(node, ast.FunctionDef):
n = node.end_lineno - node.lineno + 1
if n > 50:
over.append("%s:%s:%d" % (f.name, node.name, n))
self.assertEqual(over, [])
class PublicApiTypesTest(unittest.TestCase):
def test_public_api_fully_annotated_and_no_any(self):
import re
unannotated = []
for f in CHV.glob("*.py"):
text = f.read_text()
for line in text.splitlines():
self.assertIsNone(re.search(r"\bAny\b", line),
"%s 含 Any 逃逸:%s" % (f.name, line))
tree = ast.parse(text)
for node in tree.body:
if isinstance(node, ast.FunctionDef) \
and not node.name.startswith("_"):
bad = [a.arg for a in node.args.args
if a.annotation is None]
if bad or node.returns is None:
unannotated.append("%s:%s" % (f.name, node.name))
if isinstance(node, ast.ClassDef):
for m in node.body:
if isinstance(m, ast.FunctionDef) and (
not m.name.startswith("_")
or m.name == "__init__"):
bad = [a.arg for a in m.args.args
if a.arg != "self"
and a.annotation is None]
bad += [a.arg for a in m.args.kwonlyargs
if a.annotation is None]
if bad or m.returns is None:
unannotated.append(
"%s:%s.%s" % (f.name, node.name,
m.name))
self.assertEqual(unannotated, [])
def test_corrected_annotations_match_runtime(self):
from cloud.chv.disk import record_snapshot
from cloud.chv.hostkeys import host_key_paths
with tempfile.TemporaryDirectory() as d:
entries, removed = record_snapshot(
d, "vm1", "https://example.invalid/snap", 20.0)
self.assertIsInstance(entries, list)
self.assertIsInstance(removed, list)
paths = host_key_paths(d, "vm1")
self.assertIsInstance(paths, dict)
self.assertIn("private", paths)
class ReplacementChecklistTest(unittest.TestCase):
def test_points_resolve_and_215_modules_absent(self):
by_id = {p["id"]: p for p in REPLACEMENT_POINTS}
self.assertEqual(set(by_id), {"R1", "R2", "R3"})
self.assertIsNotNone(resolve_inside)
self.assertIsNotNone(tmpfs_isolated)
hardening = (ROOT / "cloud" / "chv" / "hardening.py").read_text()
self.assertIn("resolve_inside", hardening)
for mod in ("cloud.gates", "cloud.sandbox_policy",
"cloud.timeouts", "cloud.quota"):
self.assertIsNone(importlib.util.find_spec(mod), mod)
import cloud.chv.compat215 as m
self.assertFalse(hasattr(m, "resolve_module_timeout"))
self.assertFalse(hasattr(m, "decide"))
self.assertIsNone(by_id["R3"]["replacement"])
self.assertIsNone(by_id["R2"]["caller"])
def test_tmpfs_isolated_has_no_production_caller(self):
hits = []
for f in (ROOT / "cloud").rglob("*.py"):
if "tests" in f.parts:
continue
if "tmpfs_isolated" in f.read_text():
hits.append(str(f.relative_to(ROOT)))
self.assertEqual(hits, ["cloud/chv/compat215.py"])
def _row(**kw):
base = {
"id": "t1", "project_id": "p1", "prompt": "x",
"status": "interrupted", "idempotency_key": None,
"depends_on": "[]", "priority": 0, "created_by": "u",
"created_at": 0, "updated_at": 0, "last_heartbeat": 0,
}
base.update(kw)
return base
class ServerLifecycleTest(unittest.TestCase):
def test_goal_fallback_form_also_derives_dead(self):
row = _row(executor="cloud-hypervisor",
interrupted_reason="VM 中途死亡")
self.assertTrue(_chv_vm_dead_from_row(row))
self.assertTrue(_task_to_dict(row).get("chv_vm_dead"))
def test_non_vm_reasons_never_derive(self):
for reason in ("quota", "worker_shutdown", None, ""):
row = _row(executor="cloud-hypervisor",
interrupted_reason=reason)
self.assertIsNone(_chv_vm_dead_from_row(row))
self.assertNotIn("chv_vm_dead", _task_to_dict(row))
row = _row(executor="docker",
interrupted_reason="VM 中途死亡,任务未完成")
self.assertIsNone(_chv_vm_dead_from_row(row))
if __name__ == "__main__":
unittest.main()