"""CHV 架构 9.5+ 提升轨测试（feat-chv-arch95plus，堆叠于 #252）。 只用标准库与 mock，绝不启动真实 VM；测试里只用 FAKE 占位密钥，绝无真实密钥。覆盖：长函数提取后的行为不变、 WarmPool facade 不调实验接口、#215 对齐接口形状、审计 失败计数不中断、只读字段 chv_vm_dead 的派生规则、 lockfile 校验脚本的通过/漂移/空清单。 """ import json import logging import os import subprocess import sys import tempfile import unittest from pathlib import Path try: from cloud.chv.compat215 import resolve_inside, tmpfs_isolated from cloud.chv.hardening import validate_snapshot_url from cloud.chv.pool import reconcile_orphans_for_executor from cloud.sandbox import CloudHypervisorExecutor from cloud.server import _task_to_dict except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.chv.compat215 import resolve_inside, tmpfs_isolated from cloud.chv.hardening import validate_snapshot_url from cloud.chv.pool import reconcile_orphans_for_executor from cloud.sandbox import CloudHypervisorExecutor from cloud.server import _task_to_dict 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 FakeExecutor: def __init__(self): self.acquire_called = False self.prewarm_called = False def shutdown_vm(self, vm): return True def _emit(self, *args, **kwargs): pass class Compat215Test(unittest.TestCase): def test_resolve_inside_blocks_escape_and_symlink(self): with tempfile.TemporaryDirectory() as d: inside = os.path.join(d, "a.txt") with open(inside, "w") as fh: fh.write("x") self.assertEqual(resolve_inside(d, "a.txt"), inside) self.assertIsNone(resolve_inside(d, "../outside")) self.assertIsNone(resolve_inside(d, "/etc/passwd")) self.assertEqual(resolve_inside(d, ""), d) def test_tmpfs_isolated_chv_and_unknown_fail_closed(self): isolated, _ = tmpfs_isolated("cloud-hypervisor", "/tmp") self.assertFalse(isolated) isolated, _ = tmpfs_isolated("unknown-kind", "/tmp") self.assertFalse(isolated) def test_snapshot_url_uses_resolve_inside(self): with tempfile.TemporaryDirectory() as d: state = os.path.join(d, "state") os.makedirs(state) path = os.path.join(state, "s.raw") with open(path, "w") as fh: fh.write("x") self.assertEqual( validate_snapshot_url("file://" + path, state), path) with self.assertRaises(Exception): validate_snapshot_url( "file:///etc/passwd", state) def test_no_timeout_or_quota_map(self): import cloud.chv.compat215 as m self.assertFalse(hasattr(m, "resolve_module_timeout")) self.assertFalse(hasattr(m, "decide")) class PoolFacadeTest(unittest.TestCase): def test_facade_only_reconciles_never_acquires(self): # facade 内部构造 WarmPool：用假执行器 + 空 registry， # 只要它不调 acquire/prewarm 就返回 [] 且不炸。 ex = FakeExecutor() # WarmPool 构造需要 state_dir 属性；没有则 registry 为空 ex.state_dir = tempfile.mkdtemp() ex.warm_pool_size = 0 ex.snapshot_url = None result = reconcile_orphans_for_executor(ex) self.assertEqual(result, []) self.assertFalse(ex.acquire_called) self.assertFalse(ex.prewarm_called) class AuditCounterTest(unittest.TestCase): def test_emit_failure_counts_and_does_not_raise(self): ex = CloudHypervisorExecutor.__new__(CloudHypervisorExecutor) ex.audit_callback = lambda *a, **k: (_ for _ in ()).throw( RuntimeError("db down")) ex.audit_failure_count = 0 # _emit 不抛异常（任务不因此中断），计数 +1 ex._emit("vm_boot", {"id": "t1"}, "vm=chv-t1") self.assertEqual(ex.audit_failure_count, 1) ex._emit("vm_boot", {"id": "t1"}, "vm=chv-t1") self.assertEqual(ex.audit_failure_count, 2) class ChvVmDeadFieldTest(unittest.TestCase): def test_only_chv_with_vm_death_reason_returns_true(self): row = _row(executor="cloud-hypervisor", interrupted_reason="VM 中途死亡，任务未完成") d = _task_to_dict(row) self.assertTrue(d.get("chv_vm_dead")) row2 = _row(executor="cloud-hypervisor", interrupted_reason="quota") self.assertNotIn("chv_vm_dead", _task_to_dict(row2)) row3 = _row(executor="docker", interrupted_reason="VM 中途死亡，任务未完成") self.assertNotIn("chv_vm_dead", _task_to_dict(row3)) row4 = _row(executor="cloud-hypervisor", interrupted_reason=None) self.assertNotIn("chv_vm_dead", _task_to_dict(row4)) class LockfileScriptTest(unittest.TestCase): def _run(self, data, extra=None): with tempfile.TemporaryDirectory() as d: p = os.path.join(d, "BUILDINFO.json") with open(p, "w") as fh: json.dump(data, fh) script = str(Path(__file__).resolve().parents[2] / "chv-guest" / "check_lockfile.py") cmd = [sys.executable, script, p] + (extra or []) return subprocess.run(cmd, capture_output=True, text=True) def test_pass_and_empty_and_drift(self): good = {"tag": "t", "versions": {"dpkg": ["a 1"], "pip": ["b==2"]}} self.assertEqual(self._run(good, ["--tag", "t"]).returncode, 0) bad = {"tag": "t", "versions": {"dpkg": [], "pip": []}} self.assertNotEqual(self._run(bad).returncode, 0) with tempfile.TemporaryDirectory() as d: frozen = os.path.join(d, "frozen.json") with open(frozen, "w") as fh: json.dump({"dpkg": ["missing 9"], "pip": []}, fh) self.assertNotEqual( self._run(good, ["--frozen", frozen]).returncode, 0) class LongFunctionTest(unittest.TestCase): def test_no_chv_function_over_50_lines(self): import ast base = Path(__file__).resolve().parents[2] / "cloud" / "chv" over = [] for f in base.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, []) if __name__ == "__main__": unittest.main()