"""CHV 混合模式 C 轨测试（feat-chv-hybrid-c）。 只覆盖本轨的三个改动点（DESIGN-HYBRID.md §9.1 第 10–13 项）： - autoscaler desired_workers 作用域扩展（§1.2，W=2）； - 回收同步失败即任务标同步失败、VM 不回收（§3.3）； - 审计事件目录（§7，事件名与设计原文一致、带 host_id、 不带密钥值）与 orphans facade 核实（§1.3/ADR-5/ADR-12）。 只用标准库与 mock runner/launcher/api_client，绝不启动 真实 VM、不联网、不碰生产库。真机验证（WSL 3 台排布、 热恢复 ≤2 秒等）不在本文件声称已验证。 """ import json import os import socket import subprocess import sys import tempfile import time import unittest from pathlib import Path from unittest import mock try: from cloud.autoscaler import ( CHV_WARM_POOL_PER_HOST, Autoscaler, desired_workers, pool_target_for_hosts, ) from cloud.chv import hybrid_events from cloud.chv.executor import CloudHypervisorExecutor from cloud.chv.orphans import reconcile_orphans_for_executor from cloud.server import init_db except ImportError: sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.autoscaler import ( CHV_WARM_POOL_PER_HOST, Autoscaler, desired_workers, pool_target_for_hosts, ) from cloud.chv import hybrid_events from cloud.chv.executor import CloudHypervisorExecutor from cloud.chv.orphans import reconcile_orphans_for_executor from cloud.server import init_db class FakeResult: def __init__(self, returncode=0, stdout="", stderr=""): self.returncode = returncode self.stdout = stdout self.stderr = stderr class FakeProc: def __init__(self, alive=True): self._alive = alive self.terminated = False self.killed = False def poll(self): return None if self._alive else 1 def terminate(self): self.terminated = True self._alive = False def wait(self, timeout=None): return 0 def kill(self): self.killed = True self._alive = False def _write(path, content=b"x"): with open(path, "wb") as fh: fh.write(content) os.chmod(path, 0o755) return path # -- 11. desired_workers + 池目标数 -------------------------------------- class DesiredWorkersPoolTestCase(unittest.TestCase): def test_pool_target_zero_unchanged(self): # pool_target=0（缺省）时与旧公式一字不差。 self.assertEqual(desired_workers(0, 2, 1, 2), 1) self.assertEqual(desired_workers(0, 2, 0, 2), 0) self.assertEqual(desired_workers(1, 2, 1, 2), 1) self.assertEqual(desired_workers(3, 2, 1, 2), 2) self.assertEqual(desired_workers(10, 2, 1, 2), 2) self.assertEqual(desired_workers(5, 1, 1, 3), 3) self.assertEqual(desired_workers(5, 1, 1, 3, pool_target=0), 3) def test_pool_target_additive(self): # 期望值 = fleet 需要量 + 池目标数（§1.2）。 self.assertEqual(desired_workers(0, 2, 1, 20, pool_target=12), 13) self.assertEqual(desired_workers(4, 2, 1, 20, pool_target=12), 14) self.assertEqual(desired_workers(3, 2, 1, 20, pool_target=2), 4) # depth=0 且 min_workers=0 时池目标仍维持（预热不因 # 队列空而归零）。 self.assertEqual(desired_workers(0, 2, 0, 20, pool_target=12), 12) def test_pool_target_capped_by_max_workers(self): # 池目标数计入 max_workers 上限：配不下即截断，不静默多扩。 self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=12), 2) self.assertEqual(desired_workers(10, 2, 1, 3, pool_target=12), 3) def test_pool_target_negative_or_none_is_zero(self): self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=-5), 1) self.assertEqual(desired_workers(0, 2, 1, 2, pool_target=None), 1) class PoolTargetForHostsTestCase(unittest.TestCase): def test_w_is_two_per_host(self): self.assertEqual(CHV_WARM_POOL_PER_HOST, 2) def test_pool_target_is_w_times_online_hosts(self): self.assertEqual(pool_target_for_hosts(6), 12) self.assertEqual(pool_target_for_hosts(22), 44) self.assertEqual(pool_target_for_hosts(1), 2) self.assertEqual(pool_target_for_hosts(0), 0) def test_invalid_hosts_is_zero(self): self.assertEqual(pool_target_for_hosts(-1), 0) self.assertEqual(pool_target_for_hosts(None), 0) self.assertEqual(pool_target_for_hosts("not-a-number"), 0) def _insert_task(db_path, task_id): now = time.time() import sqlite3 conn = sqlite3.connect(db_path) try: conn.execute( "INSERT INTO tasks(id, project_id, prompt, status," " idempotency_key, depends_on, priority, created_by," " created_at, updated_at, last_heartbeat)" " VALUES(?, 'p1', 'x', 'queued', NULL, '[]', 0," " 'tester', ?, ?, NULL)", (task_id, now, now)) conn.commit() finally: conn.close() class AutoscalerPoolReconcileTestCase(unittest.TestCase): """L1 其余语义不变：30 秒一轮（interval 缺省）、scale up fast / scale down slow（scale_down_after 冷却）原样复用， 只验证期望值多了池目标数。""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.db_path = os.path.join(self.tmp.name, "control.db") init_db(self.db_path) self.state_file = os.path.join(self.tmp.name, "state.json") self.fake_cmd = [sys.executable, "-c", "import time; time.sleep(120)"] def tearDown(self): if hasattr(self, "scaler"): for wid in list(self.scaler.workers): try: self.scaler._stop_worker(wid) except Exception: pass self.tmp.cleanup() def _scaler(self, **kwargs): kwargs.setdefault("min_workers", 1) kwargs.setdefault("max_workers", 5) kwargs.setdefault("per_worker_capacity", 2) kwargs.setdefault("scale_down_after", 1) kwargs.setdefault("drain_timeout", 5) kwargs.setdefault("state_file", self.state_file) self.scaler = Autoscaler( self.db_path, [], worker_cmd=self.fake_cmd, **kwargs) return self.scaler def test_empty_queue_still_holds_pool(self): scaler = self._scaler(pool_target=2) depth, desired, actual = scaler.reconcile() self.assertEqual((depth, desired, actual), (0, 3, 3)) def test_depth_plus_pool_and_scale_down_slow(self): scaler = self._scaler(pool_target=2, scale_down_after=2) for tid in ("A", "B", "C", "D"): _insert_task(self.db_path, tid) depth, desired, actual = scaler.reconcile() self.assertEqual((depth, desired, actual), (4, 4, 4)) # 任务全完成：深度归零但池目标仍在，期望值回落到 3 # 而不是 1；缩容仍须等满 2 轮冷却（scale down slow）。 import sqlite3 conn = sqlite3.connect(self.db_path) try: conn.execute("UPDATE tasks SET status='completed'") conn.commit() finally: conn.close() scaler.reconcile() self.assertEqual(len(scaler.workers), 4) self.assertEqual(scaler.last_action, "scale_down_pending_1/2") depth, desired, actual = scaler.reconcile() self.assertEqual((depth, desired, actual), (0, 3, 3)) def test_online_hosts_derives_pool_target(self): scaler = self._scaler(online_hosts=6, max_workers=20) self.assertEqual(scaler.pool_target, 12) depth, desired, actual = scaler.reconcile() self.assertEqual((depth, desired, actual), (0, 13, 13)) def test_explicit_pool_target_wins_over_hosts(self): scaler = self._scaler(pool_target=2, online_hosts=6) self.assertEqual(scaler.pool_target, 2) def test_state_file_records_pool_target(self): scaler = self._scaler(pool_target=2) scaler.reconcile() with open(self.state_file) as fh: state = json.load(fh) self.assertEqual(state["pool_target"], 2) self.assertEqual(state["desired"], 3) def test_main_passes_pool_flags(self): import cloud.autoscaler as autoscaler_mod with mock.patch.object( autoscaler_mod, "Autoscaler") as mock_cls, \ mock.patch("signal.signal"): mock_cls.return_value.run_loop = mock.Mock() autoscaler_mod.main([ "--db", self.db_path, "--pool-target", "12", "--", "--interval", "10"]) kwargs = mock_cls.call_args.kwargs self.assertEqual(kwargs["pool_target"], 12) with mock.patch.object( autoscaler_mod, "Autoscaler") as mock_cls, \ mock.patch("signal.signal"): mock_cls.return_value.run_loop = mock.Mock() autoscaler_mod.main([ "--db", self.db_path, "--online-hosts", "6", "--", "--interval", "10"]) kwargs = mock_cls.call_args.kwargs self.assertEqual(kwargs["online_hosts"], 6) self.assertEqual(kwargs["warm_pool_per_host"], 2) # -- 12. 回收同步 --------------------------------------------------------- class RecycleSyncCase(unittest.TestCase): """mock runner/launcher/api_client，不启动真实 VM。 push_rc 控制客内 `chv-sync push` 的返回值。""" TOKEN = "SECRET-TOKEN-XYZ-DO-NOT-LEAK" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.dir = self.tmp.name self.binary = _write(os.path.join(self.dir, "cloud-hypervisor")) self.ch_remote = _write(os.path.join(self.dir, "ch-remote")) self.kernel = _write(os.path.join(self.dir, "vmlinux")) self.rootfs = _write(os.path.join(self.dir, "disk.raw"), b"R" * 100) self.ssh_key = _write(os.path.join(self.dir, "id_chv")) self.kvm = _write(os.path.join(self.dir, "kvm")) self.cpuinfo = os.path.join(self.dir, "cpuinfo") with open(self.cpuinfo, "w") as fh: fh.write("flags : fpu vmx sse\n") self.state_dir = os.path.join(self.dir, "state") self.calls = [] self.audit = [] self.proc = FakeProc(alive=True) self.push_rc = 0 self.push_stdout = "已 push" self.push_stderr = "" self.raise_on_push = False def tearDown(self): self.tmp.cleanup() def _runner(self, cmd, timeout=None): self.calls.append(list(cmd)) joined = " ".join(str(c) for c in cmd) if "chv-sync" in joined: if self.raise_on_push: raise RuntimeError("ssh broken") return FakeResult(self.push_rc, self.push_stdout, self.push_stderr) return FakeResult(0, "", "") def _make_exec(self, sync_configured=True): env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", "CHV_HOST_KEY_MODE": "keyscan", "CHV_SSH_FORWARD": "1"} if sync_configured: env.update({ "CHV_GITEA_URL": "https://gitea.tose.top", "CHV_GITEA_USER": "tester", "CHV_GITEA_TOKEN": self.TOKEN, "CHV_SYNC_REPO": "tester/repo", }) ex = CloudHypervisorExecutor( binary=self.binary, ch_remote=self.ch_remote, kernel=self.kernel, rootfs=self.rootfs, ssh_key=self.ssh_key, kvm_path=self.kvm, cpuinfo_path=self.cpuinfo, state_dir=self.state_dir, runner=self._runner, launcher=lambda cmd: self.proc, api_client=lambda *a, **k: (200, {}), sleeper=lambda s: None, env=env) ex.audit_callback = lambda e, t, d, k=None: self.audit.append( (e, t.get("id") if isinstance(t, dict) else t, d, k)) return ex def _task(self): return {"id": "abc123", "prompt": "x", "project_id": "proj1"} def _mark_vm_running(self, ex, vm_name="chv-abc123"): os.makedirs(self.state_dir, exist_ok=True) with open(os.path.join(self.state_dir, vm_name + ".sock"), "w") as fh: fh.write("") ex._vms[vm_name] = self.proc def _audit_events(self): return [e for e, _t, _d, _k in self.audit] def _audit_details(self): return [d for _e, _t, d, _k in self.audit] def test_no_sync_configured_recycles_as_before(self): # 未配 Gitea 同步：无物可推，走旧回收路径（回归）。 ex = self._make_exec(sync_configured=False) self._mark_vm_running(ex) out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "recycled") self.assertTrue(out["sync_ok"]) self.assertFalse(out["vm_retained"]) self.assertTrue(self.proc.terminated) self.assertIn("vm_recycled", self._audit_events()) self.assertFalse(any("chv-sync" in " ".join(c) for c in self.calls)) def test_sync_push_ok_then_recycle(self): ex = self._make_exec() self._mark_vm_running(ex) self.push_rc = 0 out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "recycled") self.assertTrue(out["sync_ok"]) self.assertFalse(out["vm_retained"]) self.assertTrue(self.proc.terminated) self.assertIn("vm_recycled", self._audit_events()) self.assertTrue(any("chv-sync" in " ".join(c) for c in self.calls)) self.assertNotIn("vm_recycle_sync_failed", self._audit_events()) def test_sync_push_failed_marks_sync_failed_and_retains_vm(self): # §3.3 新增语义：同步失败即任务标同步失败、VM 不回收。 ex = self._make_exec() self._mark_vm_running(ex) self.push_rc = 1 self.push_stderr = "push 被拒绝（远端已有新提交）" out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "sync_failed") self.assertEqual(out["status"], "sync_failed") self.assertFalse(out["sync_ok"]) self.assertTrue(out["vm_retained"]) self.assertIn("push", out["sync_error"]) # VM 保留：进程未停、socket 未删、未发 recycled/shutdown。 self.assertFalse(self.proc.terminated) self.assertTrue(os.path.exists(os.path.join( self.state_dir, "chv-abc123.sock"))) self.assertNotIn("vm_recycled", self._audit_events()) self.assertNotIn("vm_shutdown", self._audit_events()) self.assertNotIn("vm_destroy", self._audit_events()) # 记审计：事件名正确、带 host_id、不带密钥值。 self.assertIn("vm_recycle_sync_failed", self._audit_events()) details = " ".join(self._audit_details()) self.assertIn("host_id=", details) self.assertNotIn(self.TOKEN, details) self.assertNotIn(self.TOKEN, json.dumps(self.audit)) def test_sync_ssh_error_also_retains_vm(self): ex = self._make_exec() self._mark_vm_running(ex) self.raise_on_push = True out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "sync_failed") self.assertTrue(out["vm_retained"]) self.assertFalse(self.proc.terminated) self.assertIn("vm_recycle_sync_failed", self._audit_events()) def test_sync_failed_then_retry_after_push_ok(self): # 人工/重试路径：push 修好后再次回收即成功，不卡死。 ex = self._make_exec() self._mark_vm_running(ex) self.push_rc = 1 out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "sync_failed") self.push_rc = 0 out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "recycled") self.assertTrue(self.proc.terminated) def test_sync_skip_flag_recycles_despite_push_failure(self): ex = self._make_exec() self._mark_vm_running(ex) self.push_rc = 1 out = ex.recycle_vm(self._task(), sync=False) self.assertEqual(out["state"], "recycled") self.assertTrue(self.proc.terminated) def test_vm_not_running_does_not_block_cleanup(self): # VM 已死：没有活工作树可推，不拿死 VM 卡清理。 ex = self._make_exec() out = ex.recycle_vm(self._task()) self.assertEqual(out["state"], "recycled") def test_recycle_persistent_vm_sync_failed_keeps_disks(self): ex = self._make_exec() self._mark_vm_running(ex, vm_name="chv-persist-proj1") self.push_rc = 1 out = ex.recycle_persistent_vm({"id": "chv-persist-proj1"}) self.assertEqual(out["state"], "sync_failed") self.assertEqual(out["status"], "sync_failed") self.assertTrue(out["vm_retained"]) self.assertEqual(out["deleted_disks"], []) self.assertFalse(self.proc.terminated) self.assertNotIn("vm_persistent_recycled", self._audit_events()) self.assertIn("vm_recycle_sync_failed", self._audit_events()) def test_recycle_vm_missing_id_still_raises(self): ex = self._make_exec() with self.assertRaises(ValueError): ex.recycle_vm({}) # -- 13. 审计事件目录 ------------------------------------------------------ class HybridEventsTestCase(unittest.TestCase): def test_design_event_names_verbatim(self): # 事件名与 DESIGN-HYBRID.md §9.1-13 原文一字不差。 self.assertEqual(hybrid_events.HYBRID_EVENTS, ( "warm_pool_acquire", "warm_pool_miss", "warm_pool_release", "vm_resident_suspended", "vm_resident_demoted", "vm_resident_promoted", "resident_upgrade_hint", "snapshot_tenant_mismatch", "vm_restore_denied", )) def test_every_event_requires_host_id(self): for event in hybrid_events.HYBRID_EVENTS: self.assertIn("host_id", hybrid_events.REQUIRED_FIELDS[event], event) self.assertIn( "host_id", hybrid_events.REQUIRED_FIELDS[ hybrid_events.VM_RECYCLE_SYNC_FAILED]) def test_build_event_defaults_host_id(self): fields = hybrid_events.build_event( "warm_pool_miss", {"tenant_id": "t1"}) self.assertEqual(fields["host_id"], socket.gethostname()) self.assertEqual(fields["tenant_id"], "t1") def test_build_event_rejects_missing_required(self): with self.assertRaises(ValueError): hybrid_events.build_event("warm_pool_acquire", {}) with self.assertRaises(ValueError): hybrid_events.build_event( "vm_resident_suspended", {"host_id": "h1", "tenant_id": "t1"}) with self.assertRaises(ValueError): hybrid_events.build_event("no_such_event", {"host_id": "h1"}) def test_secret_field_rejected_without_leaking_value(self): sentinel = "SECRET-TOKEN-XYZ-DO-NOT-LEAK" with self.assertRaises(ValueError) as ctx: hybrid_events.build_event( "warm_pool_acquire", {"vm": "chv-x", "GITEA_TOKEN": sentinel}) self.assertNotIn(sentinel, str(ctx.exception)) self.assertIn("GITEA_TOKEN", str(ctx.exception)) def test_format_detail_has_host_id_and_no_secret(self): detail = hybrid_events.format_detail( "vm_resident_promoted", {"tenant_id": "t1", "vm": "chv-persist-t1"}) self.assertIn("host_id=%s" % socket.gethostname(), detail) self.assertIn("vm=chv-persist-t1", detail) self.assertIn("tenant_id=t1", detail) # -- 10. orphans facade 核实 ---------------------------------------------- class OrphansFacadeTestCase(unittest.TestCase): def test_facade_exists_and_cleans_only_orphans(self): # facade 已存在：读 registry → 只清理 warm/assigned # 条目（只清理、不分配），非法条目丢弃不连坐。 with tempfile.TemporaryDirectory() as tmp: registry = { "warm-1": {"state": "warm", "vm": "chv-a", "task_id": "t1"}, "asg-1": {"state": "assigned", "vm": "chv-b", "task_id": "t2"}, "bad-1": {"state": "warm"}, "dead-1": {"state": "recycled", "vm": "chv-c"}, } with open(os.path.join(tmp, "warm-pool.json"), "w") as fh: json.dump(registry, fh) class Exec: state_dir = tmp cleaned = [] def shutdown_vm(self, vm): self.cleaned.append(vm) return True def _emit(self, event, task, detail, kind=None): pass ex = Exec() cleaned = reconcile_orphans_for_executor(ex) self.assertEqual(cleaned, ["chv-a", "chv-b"]) self.assertEqual(ex.cleaned, ["chv-a", "chv-b"]) with open(os.path.join(tmp, "warm-pool.json")) as fh: self.assertEqual(json.load(fh), {}) def test_facade_quarantines_corrupt_registry(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "warm-pool.json") with open(path, "w") as fh: fh.write("{not json") class Exec: state_dir = tmp def shutdown_vm(self, vm): raise AssertionError("must not be called") self.assertEqual(reconcile_orphans_for_executor(Exec()), []) self.assertTrue(os.path.exists(path + ".corrupt")) def test_warm_pool_still_not_wired_to_dispatch(self): from cloud.chv.experimental_pool import WarmPool self.assertIs(WarmPool.wired_to_dispatch, False) def test_orphans_module_does_not_import_experimental_pool(self): # 生产 facade 不 import 实验模块（ADR-12）：全新解释器 # 里只 import orphans，experimental_pool 不得被加载。 code = ( "import cloud.chv.orphans, sys;" " assert 'cloud.chv.experimental_pool' not in sys.modules;" " print('orphans-clean')" ) proc = subprocess.run( [sys.executable, "-c", code], capture_output=True, text=True, cwd=str( Path(__file__).resolve().parents[2])) self.assertEqual(proc.returncode, 0, proc.stderr) self.assertIn("orphans-clean", proc.stdout) if __name__ == "__main__": unittest.main()