"""CHV 混合模式 A 轨（feat-chv-hybrid-a）测试。 只用标准库与 mock runner/launcher/api_client，绝不启动真实 VM（与 test_chv_opt_stage1.py 同款边界）。覆盖 DESIGN-HYBRID.md §9.1 第 1–4 项： - 混合路由分流（常驻 / 池命中 / 池空冷启 / 排队拒绝） - 干净池预热不带租户（无数据盘、无 seed、无同步等待） - 数据盘挂载归属 + “任务盘来源”不变量 - snapshot_tenant_mismatch（含无 tenant_id 登记拒绝） - 已定值常量（每宿主预热 2 台、池上限 50%、常驻上限 50%、 排队超时 300 秒、热恢复 ≤2 秒、冷恢复 ≤30 秒、快照 1 个/30 天） 不改其它测试文件；阶段 1 的旧语义（prewarm_persistent_vm / acquire_for_task / dispatch_task_persistent / _validate_restore 的 legacy 警告放行）原样保留，只测本轨新增行为。 """ import json import os import sys import tempfile import unittest from pathlib import Path from unittest import mock try: from cloud.chv import ChvFailure from cloud.chv import CloudHypervisorExecutor from cloud.chv.disk import ( read_persistent_vm, read_vm_tenant, record_snapshot, write_persistent_vm, ) from cloud.chv.executor import ( HYBRID_COLD_RESTORE_TARGET_SECONDS, HYBRID_QUEUE_TIMEOUT_SECONDS, HYBRID_SNAPSHOT_PER_TENANT, HYBRID_SNAPSHOT_RETENTION_DAYS, HYBRID_WARM_POOL_PER_HOST, HYBRID_WARM_RESUME_TARGET_SECONDS, ) except ImportError: # discover -s cloud/tests 时把项目根目录补进 sys.path sys.path.insert(0, str(Path(__file__).resolve().parents[2])) from cloud.chv import ChvFailure from cloud.chv import CloudHypervisorExecutor from cloud.chv.disk import ( read_persistent_vm, read_vm_tenant, record_snapshot, write_persistent_vm, ) from cloud.chv.executor import ( HYBRID_COLD_RESTORE_TARGET_SECONDS, HYBRID_QUEUE_TIMEOUT_SECONDS, HYBRID_SNAPSHOT_PER_TENANT, HYBRID_SNAPSHOT_RETENTION_DAYS, HYBRID_WARM_POOL_PER_HOST, HYBRID_WARM_RESUME_TARGET_SECONDS, ) 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 def poll(self): return None if self._alive else 1 def terminate(self): self._alive = False def wait(self, timeout=None): return 0 def kill(self): self._alive = False def _write(path, content=b"x"): with open(path, "wb") as fh: fh.write(content) os.chmod(path, 0o755) return path class HybridACase(unittest.TestCase): 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, "base.qcow2"), 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.api_calls = [] self.launches = [] self.vm_states = {} self.events = [] def tearDown(self): self.tmp.cleanup() def make_exec(self, runner=None, api_client=None, launcher=None, env_extra=None, **overrides): env = {"CHV_EGRESS_PROXY": "http://10.0.0.1:8888", "CHV_HOST_KEY_MODE": "keyscan", "CHV_SSH_FORWARD": "1", "CHV_WARM_POOL_SIZE": "2"} if env_extra: env.update(env_extra) if runner is None: runner = self._default_runner if api_client is None: api_client = self._default_api if launcher is None: launcher = self._default_launcher kwargs = dict( 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=runner, launcher=launcher, api_client=api_client, sleeper=lambda s: None, env=env) kwargs.update(overrides) ex = CloudHypervisorExecutor(**kwargs) ex.audit_callback = lambda e, t, d, k: self.events.append( (e, d, k)) return ex def _default_runner(self, cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh-keyscan": return FakeResult( 0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "") if cmd[0] == "ssh": if cmd[-1] == "true": return FakeResult(0, "", "") if "dispatch" in cmd: return FakeResult(0, json.dumps( {"status": "completed", "text": "ok-dispatch"}), "") return FakeResult(0, "", "") return FakeResult(0, "", "") def _default_api(self, socket_path, method, path, payload=None): self.api_calls.append((method, path, payload)) if path == "/vm.info": state = self.vm_states.get( os.path.basename(socket_path)[:-len(".sock")]) if state: return (200, {"state": state}) return (200, {}) return (200, {}) def _default_launcher(self, cmd): self.launches.append(list(cmd)) return FakeProc(alive=True) def _task(self, task_id="abc123", project="proj1", tier=None): task = {"id": task_id, "prompt": "做一个加法函数", "project_id": project} if tier is not None: task["tier"] = tier return task def _socket(self, vm): os.makedirs(self.state_dir, exist_ok=True) path = os.path.join(self.state_dir, vm + ".sock") with open(path, "w") as fh: fh.write("") return path def _event_names(self): return [e for e, _d, _k in self.events] def _event_details(self, name): return [d for e, d, _k in self.events if e == name] # -- 已定值常量 ------------------------------------------------------------- class SettledValuesTest(HybridACase): def test_hybrid_settled_values(self): ex = self.make_exec() self.assertEqual(HYBRID_WARM_POOL_PER_HOST, 2) self.assertEqual(ex.warm_pool_per_host_target(), 2) self.assertEqual(ex.clean_pool_target(), 2) # 池上限 = 低频并发峰值的 50%：20 → 10，50 → 25。 self.assertEqual(ex.warm_pool_cap(20), 10) self.assertEqual(ex.warm_pool_cap(50), 25) self.assertIsNone(ex.warm_pool_cap(None)) # 生效目标取 min（配置目标，上限）：目标 2、峰值 20 → 2。 self.assertEqual(ex.clean_pool_effective_target(20), 2) ex_big = self.make_exec(env_extra={ "CHV_WARM_POOL_SIZE": "30", "CHV_LOW_FREQ_PEAK_CONCURRENCY": "20"}) self.assertEqual(ex_big.clean_pool_effective_target(), 10) # 常驻上限 = 槽位 50%：4 槽 → 2 台，48 槽 → 24 台。 self.assertEqual(ex.resident_cap(), 2) ex48 = self.make_exec(env_extra={"CHV_MAX_VMS": "48"}) self.assertEqual(ex48.resident_cap(), 24) # 排队超时 300 秒；热恢复 ≤2 秒、冷快照恢复 ≤30 秒（目标值， # 只作常量/注释，不在单元测试里硬测时延）；快照 1 个/30 天。 self.assertEqual(HYBRID_QUEUE_TIMEOUT_SECONDS, 300) self.assertEqual(HYBRID_WARM_RESUME_TARGET_SECONDS, 2) self.assertEqual(HYBRID_COLD_RESTORE_TARGET_SECONDS, 30) self.assertEqual(HYBRID_SNAPSHOT_PER_TENANT, 1) self.assertEqual(HYBRID_SNAPSHOT_RETENTION_DAYS, 30) # -- 混合路由分流 ----------------------------------------------------------- class HybridRouteTest(HybridACase): def test_route_by_tier(self): ex = self.make_exec() self.assertEqual( ex._hybrid_route(self._task(tier="paid")), "resident") self.assertEqual( ex._hybrid_route(self._task(tier="heavy")), "resident") self.assertEqual( ex._hybrid_route(self._task(tier="PAID")), "resident") self.assertEqual( ex._hybrid_route(self._task(tier="free")), "pool") self.assertEqual(ex._hybrid_route(self._task()), "pool") # 常驻是权益门：Paid 但无 tenant（无归属）即 fail-closed 走池。 task = {"id": "x1", "prompt": "p", "tier": "paid"} self.assertEqual(ex._hybrid_route(task), "pool") def test_dispatch_hybrid_validates_task(self): ex = self.make_exec() with self.assertRaises(ValueError): ex.dispatch_task_hybrid({"prompt": "p"}) with self.assertRaises(ValueError): ex.dispatch_task_hybrid({"id": "x"}) def test_dispatch_hybrid_pool_disabled_falls_back(self): ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) out = ex.dispatch_task_hybrid(self._task()) self.assertEqual(out["status"], "completed") # 退化为阶段 1 旧语义：结果不带 route/allocation。 self.assertNotIn("route", out) self.assertEqual(ex._vms, {}) def test_dispatch_pool_cold_when_pool_empty(self): ex = self.make_exec() out = ex.dispatch_task_hybrid(self._task(task_id="t1")) self.assertEqual(out["status"], "completed") self.assertEqual(out["route"], "pool") self.assertEqual(out["allocation"], "pool-cold") self.assertEqual(out["vm_mode"], "cold") self.assertIn("warm_pool_miss", self._event_names()) self.assertEqual(ex._vms, {}) def test_dispatch_pool_queue_timeout_refuses(self): ex = self.make_exec(env_extra={ "CHV_QUEUE_TIMEOUT": "0", "CHV_MAX_VMS": "1"}) # 宿主已满（1 台活 VM 占满唯一槽位），池空 → 排队 0 秒即拒绝。 ex._vms["chv-busy"] = FakeProc(alive=True) with self.assertRaises(ChvFailure) as ctx: ex.dispatch_task_hybrid(self._task(task_id="t1")) self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") self.assertEqual(self.launches, []) self.assertIn("warm_pool_miss", self._event_names()) def test_dispatch_resident_cold_when_no_resident_vm(self): ex = self.make_exec() out = ex.dispatch_task_hybrid( self._task(task_id="t1", project="paid1", tier="paid")) self.assertEqual(out["status"], "completed") self.assertEqual(out["route"], "resident") self.assertEqual(out["allocation"], "resident-cold") self.assertEqual(out["vm"], "chv-persist-paid1") rec = read_persistent_vm(self.state_dir, "paid1") self.assertEqual(rec["state"], "resident-active") self.assertEqual(rec["tier"], "paid") self.assertTrue(rec["resident"]) self.assertEqual(ex._resident_count(), 1) def test_dispatch_resident_resumed_from_prewarm(self): ex = self.make_exec() ex.prewarm_persistent_vm("paid1") self._socket("chv-persist-paid1") self.vm_states["chv-persist-paid1"] = "Paused" out = ex.dispatch_task_hybrid( self._task(task_id="t1", project="paid1", tier="paid")) self.assertEqual(out["status"], "completed") self.assertEqual(out["route"], "resident") self.assertEqual(out["allocation"], "resident-resumed") self.assertEqual(out["vm"], "chv-persist-paid1") self.assertIn("resume_seconds", out) rec = read_persistent_vm(self.state_dir, "paid1") self.assertEqual(rec["state"], "resident-active") self.assertEqual(rec["tier"], "paid") self.assertTrue(rec["resident"]) def test_resident_cap_blocks_new_resident(self): ex = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"}) # 4 槽位 → 常驻上限 2 台：已有 2 个常驻租户，第 3 个排队即拒。 for tenant in ("r1", "r2"): write_persistent_vm(self.state_dir, tenant, { "vm": "chv-persist-%s" % tenant, "tenant_id": tenant, "state": "resident-active", "origin": "task", "gitea_synced": False, "tier": "paid", "resident": True}) self.assertEqual(ex._resident_count(), 2) with self.assertRaises(ChvFailure) as ctx: ex.dispatch_task_hybrid( self._task(task_id="t1", project="r3", tier="paid")) self.assertEqual(ctx.exception.kind, "host_capacity_exceeded") self.assertEqual(self.launches, []) # 已有常驻的租户不受上限挡（自己的槽位）。 ex2 = self.make_exec(env_extra={"CHV_QUEUE_TIMEOUT": "0"}) write_persistent_vm(self.state_dir, "r1", { "vm": "chv-persist-r1", "tenant_id": "r1", "state": "resident-active", "origin": "task", "gitea_synced": False, "tier": "paid", "resident": True}) write_persistent_vm(self.state_dir, "r2", { "vm": "chv-persist-r2", "tenant_id": "r2", "state": "resident-active", "origin": "task", "gitea_synced": False, "tier": "paid", "resident": True}) self._socket("chv-persist-r1") self.vm_states["chv-persist-r1"] = "Paused" out = ex2.dispatch_task_hybrid( self._task(task_id="t1", project="r1", tier="paid")) self.assertEqual(out["route"], "resident") self.assertEqual(out["allocation"], "resident-resumed") # -- 干净池预热 ------------------------------------------------------------- class CleanPoolPrewarmTest(HybridACase): def test_prewarm_clean_pool_has_no_tenant_state(self): ex = self.make_exec() out = ex.prewarm_clean_pool() self.assertTrue(out["pooled"]) self.assertEqual(out["mode"], "prewarmed") self.assertEqual(out["vm"], "chv-pool-0") self.assertFalse(out["gitea_synced"]) self.assertTrue(out["clean"]) self.assertIsNone(out["tenant_id"]) self.assertEqual(len(self.launches), 1) vm_dir = os.path.join(self.state_dir, "chv-pool-0") # 出生时不带租户：无数据盘、无 tenant 标记、无 seed。 self.assertFalse(os.path.exists( os.path.join(vm_dir, "data.raw"))) self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0")) self.assertFalse(os.path.exists( os.path.join(vm_dir, "seed.iso"))) self.assertFalse(os.path.exists( os.path.join(vm_dir, "seed"))) # 无 Gitea 同步等待：预热全程没有任何 ssh test -d 调用。 self.assertFalse(any( "test" in call for call in self.calls if call and call[0] == "ssh")) # 登记同样不带租户。 record = ex._read_clean_pool_record(0) self.assertEqual(record["state"], "paused") self.assertIsNone(record["tenant_id"]) self.assertFalse(record["gitea_synced"]) # socket 造出来后第二次即 already-warm，不重建。 self._socket("chv-pool-0") second = ex.prewarm_clean_pool() self.assertEqual(second["mode"], "already-warm") self.assertEqual(len(self.launches), 1) def test_prewarm_clean_pool_slots_and_range(self): ex = self.make_exec() first = ex.prewarm_clean_pool(slot=1) self.assertEqual(first["vm"], "chv-pool-1") out = ex.prewarm_clean_pool(slot=5) self.assertFalse(out["pooled"]) self.assertEqual(out["reason"], "clean_pool_slot_out_of_range") self.assertEqual(len(self.launches), 1) def test_prewarm_clean_pool_disabled_when_size_zero(self): ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) self.assertEqual( ex.prewarm_clean_pool()["reason"], "warm_pool_disabled") self.assertEqual( ex.acquire_clean_pool(self._task())["mode"], "cold") self.assertEqual(ex.refill_clean_pool(), {"target": 0, "slots": []}) def test_prewarm_clean_pool_refuses_tenant_seed(self): ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.prewarm_clean_pool( extra_env={"GITEA_TOKEN": "secret-value"}) self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused") with self.assertRaises(ChvFailure) as ctx: ex.prewarm_clean_pool( extra_env={"GITEA_URL": "https://gitea.tose.top"}) self.assertEqual(ctx.exception.kind, "clean_pool_seed_refused") self.assertEqual(self.launches, []) def test_prewarm_clean_pool_never_rebuilds_running_slot(self): ex = self.make_exec() ex._write_clean_pool_record(0, { "vm": "chv-pool-0", "slot": 0, "tenant_id": "tenant-a", "state": "running", "origin": "clean-prewarm", "gitea_synced": False, "data_disk": "x", "updated_at": 0}) out = ex.prewarm_clean_pool(slot=0) self.assertFalse(out["pooled"]) self.assertEqual(out["reason"], "clean_pool_slot_busy") self.assertEqual(self.launches, []) # 自动选槽跳过正跑的 slot 0，建 slot 1。 out2 = ex.prewarm_clean_pool() self.assertEqual(out2["vm"], "chv-pool-1") self.assertEqual(len(self.launches), 1) def test_refill_clean_pool(self): ex = self.make_exec() ex.prewarm_clean_pool(slot=0) self._socket("chv-pool-0") out = ex.refill_clean_pool() self.assertEqual(out["target"], 2) self.assertEqual(out["slots"][0]["mode"], "already-warm") self.assertEqual(out["slots"][1]["mode"], "prewarmed") self.assertEqual(len(self.launches), 2) # -- 数据盘挂载归属 + 任务盘来源不变量 --------------------------------------- class DataDiskAttachTest(HybridACase): def test_verify_task_disk_source(self): ex = self.make_exec() tenant = "tenant-a" disk = ex._provision_tenant_data_disk(tenant) self.assertEqual(disk, ex._tenant_data_disk_path(tenant)) self.assertTrue(os.path.exists(disk)) # 合法的租户数据盘通过。 ex._verify_task_disk_source("chv-pool-0", tenant, disk) # 干净池 VM 自带的盘（干净快照盘）绝不能当任务盘。 pool_data = os.path.join(self.state_dir, "chv-pool-0", "data.raw") with self.assertRaises(ChvFailure) as ctx: ex._verify_task_disk_source("chv-pool-0", tenant, pool_data) self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") # 基础镜像不能当任务盘。 with self.assertRaises(ChvFailure) as ctx: ex._verify_task_disk_source("chv-pool-0", tenant, self.rootfs) self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") # 其它租户的盘不能当本租户的任务盘。 other = ex._provision_tenant_data_disk("tenant-b") with self.assertRaises(ChvFailure) as ctx: ex._verify_task_disk_source("chv-pool-0", tenant, other) self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") # 无盘 / 无租户同样拒绝。 with self.assertRaises(ChvFailure) as ctx: ex._verify_task_disk_source("chv-pool-0", tenant, None) self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") with self.assertRaises(ChvFailure) as ctx: ex._verify_task_disk_source("chv-pool-0", None, disk) self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") def test_attach_refuses_wrong_disk_before_api_call(self): ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0", data_disk=self.rootfs) self.assertEqual(ctx.exception.kind, "task_disk_source_invalid") # 不变量在 API 调用之前拦住：一次 add-disk 都没发。 self.assertFalse(any(p == "/vm.add-disk" for _m, p, _pl in self.api_calls)) def test_attach_tenant_data_disk(self): ex = self.make_exec() out = ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0") self.assertEqual(out["tenant_id"], "tenant-a") self.assertEqual(out["data_disk"], ex._tenant_data_disk_path("tenant-a")) self.assertTrue(any(p == "/vm.add-disk" for _m, p, _pl in self.api_calls)) _method, _path, payload = next( (m, p, pl) for m, p, pl in self.api_calls if p == "/vm.add-disk") self.assertEqual(payload["path"], out["data_disk"]) # 客内挂载脚本被调过（chv-data-mount）。 self.assertTrue(any( "/usr/local/bin/chv-data-mount" in call for call in self.calls if call and call[0] == "ssh")) self.assertEqual( read_vm_tenant(self.state_dir, "chv-pool-0"), "tenant-a") def test_dispatch_pool_hit_binds_tenant_disk_and_destroys_vm(self): ex = self.make_exec() ex.prewarm_clean_pool(slot=0) self._socket("chv-pool-0") self.vm_states["chv-pool-0"] = "Paused" out = ex.dispatch_task_hybrid( self._task(task_id="t1", project="tenant-a")) self.assertEqual(out["status"], "completed") self.assertEqual(out["route"], "pool") self.assertEqual(out["allocation"], "pool-hit") self.assertEqual(out["vm"], "chv-pool-0") self.assertTrue(out["warm_hit"]) self.assertEqual(out["data_disk"], ex._tenant_data_disk_path("tenant-a")) self.assertIn("resume_seconds", out) self.assertIn("warm_pool_acquire", self._event_names()) self.assertIn("warm_pool_release", self._event_names()) # 用过的池 VM 已销毁、不回池：登记清掉、VMM 登记空、归属清掉。 self.assertIsNone(ex._read_clean_pool_record(0)) self.assertEqual(ex._vms, {}) self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0")) # 租户数据盘还在（用户数据不随 VM 销毁而删）。 self.assertTrue(os.path.exists( ex._tenant_data_disk_path("tenant-a"))) # 池 VM 自带的干净盘从未被当成任务盘建出来。 self.assertFalse(os.path.exists(os.path.join( self.state_dir, "chv-pool-0", "data.raw"))) def test_acquire_sync_incomplete_fails_closed_at_allocate(self): # 分配时同步等不到：fail-closed，VM 解绑销毁，不派任务。 def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh-keyscan": return FakeResult( 0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "") if cmd[0] == "ssh": if "test" in cmd: return FakeResult(1, "", "") if cmd[-1] == "true": return FakeResult(0, "", "") if "dispatch" in cmd: return FakeResult(0, json.dumps( {"status": "completed", "text": "ok"}), "") return FakeResult(0, "", "") return FakeResult(0, "", "") ex = self.make_exec(runner=runner) ex.prewarm_clean_pool(slot=0) self._socket("chv-pool-0") self.vm_states["chv-pool-0"] = "Paused" task = self._task(task_id="t1", project="tenant-a") task["sync_workspace"] = "/home/chv/workspace/repo" with self.assertRaises(ChvFailure) as ctx: ex.dispatch_task_hybrid(task) self.assertEqual(ctx.exception.kind, "warm_pool_sync_incomplete") # 已绑过租户态的 VM 绝不回干净池：已销毁、登记清掉。 self.assertIsNone(ex._read_clean_pool_record(0)) self.assertEqual(ex._vms, {}) self.assertFalse(any("dispatch" in call for call in self.calls if call and call[0] == "ssh")) def test_release_sync_failure_keeps_vm(self): # 回收时 chv-sync push 失败：VM 不回收、不复用。 def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd[0] == "ssh-keyscan": return FakeResult( 0, "[127.0.0.1]:2222 ssh-ed25519 AAAAB3test\n", "") if cmd[0] == "ssh": if "chv-sync" in cmd: return FakeResult(1, "", "push 被拒绝") if cmd[-1] == "true": return FakeResult(0, "", "") return FakeResult(0, "", "") return FakeResult(0, "", "") ex = self.make_exec(runner=runner) ex.prewarm_clean_pool(slot=0) self._socket("chv-pool-0") self.vm_states["chv-pool-0"] = "Paused" acquired = ex.acquire_clean_pool(self._task(project="tenant-a")) self.assertEqual(acquired["mode"], "resumed") out = ex.release_pool_vm( self._task(project="tenant-a"), "chv-pool-0", sync_workspace="/home/chv/workspace/repo") self.assertFalse(out["released"]) self.assertEqual(out["reason"], "sync_failed") self.assertIn("warm_pool_release_failed", self._event_names()) # VM 还在（未销毁），登记仍是 running 绑定态。 self.assertIn("chv-pool-0", ex._vms) record = ex._read_clean_pool_record(0) self.assertEqual(record["state"], "running") self.assertEqual(record["tenant_id"], "tenant-a") # -- 快照 tenant 校验 -------------------------------------------------------- class SnapshotTenantStrictTest(HybridACase): def _snap_url(self, vm, name="snap1"): snap_dir = os.path.join(self.state_dir, vm, name) os.makedirs(snap_dir, exist_ok=True) with open(os.path.join(snap_dir, "config.json"), "w") as fh: fh.write("{}") with open(os.path.join(snap_dir, "state.json"), "w") as fh: fh.write("{}") return "file://" + snap_dir def _record(self, vm, url, tenant): record_snapshot(self.state_dir, vm, url, 20.0, tenant_id=tenant) def test_unmarked_snapshot_denied_in_hybrid_mode(self): url = self._snap_url("chv-a") self._record("chv-a", url, None) ex = self.make_exec() # CHV_WARM_POOL_SIZE=2 → 严格校验生效 with self.assertRaises(ChvFailure) as ctx: ex.restore_vm({"id": "task-a", "project_id": "tenant-a"}, url) self.assertEqual(ctx.exception.kind, "snapshot_tenant_mismatch") denied = [d for _e, d, _k in self.events if _e == "vm_restore_denied"] self.assertTrue(denied) # 审计带 host_id + 请求租户，不带快照内容（URL 不得出 # 现在 detail 里）。 self.assertIn("host_id=", denied[0]) self.assertIn("tenant=tenant-a", denied[0]) self.assertNotIn(url, denied[0]) self.assertNotIn("snap1", denied[0]) def test_unregistered_snapshot_denied_in_hybrid_mode(self): url = self._snap_url("chv-a") ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.restore_vm({"id": "task-a", "project_id": "tenant-a"}, url) self.assertEqual(ctx.exception.kind, "snapshot_tenant_mismatch") self.assertIn("vm_restore_denied", self._event_names()) def test_mismatched_snapshot_denied_in_hybrid_mode(self): url = self._snap_url("chv-a") self._record("chv-a", url, "tenant-a") ex = self.make_exec() with self.assertRaises(ChvFailure) as ctx: ex.restore_vm({"id": "task-b", "project_id": "tenant-b"}, url) self.assertEqual(ctx.exception.kind, "snapshot_tenant_mismatch") self.assertIn("vm_restore_denied", self._event_names()) def test_same_tenant_snapshot_restores_in_hybrid_mode(self): url = self._snap_url("chv-a") self._record("chv-a", url, "tenant-a") ex = self.make_exec() out = ex.restore_vm( {"id": "task-a", "project_id": "tenant-a"}, url) self.assertEqual(out["mode"], "restored") self.assertEqual(out["state"], "running") def test_explicit_strict_flag_denies_unmarked(self): url = self._snap_url("chv-a") self._record("chv-a", url, None) ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) # 显式 strict_tenant=True 时即使池未启用也拒绝。 with self.assertRaises(ChvFailure) as ctx: ex._validate_restore( {"id": "task-a", "project_id": "tenant-a"}, url, strict_tenant=True) self.assertEqual(ctx.exception.kind, "snapshot_tenant_mismatch") self.assertIn("vm_restore_denied", self._event_names()) def test_legacy_warn_pass_when_strict_opted_out(self): # 严格校验默认开启（设计 §2.2，集成时翻为默认）； # legacy 警告放行只在显式 CHV_SNAPSHOT_STRICT_TENANT=0 # 时保留（有未标记旧快照的环境的逃生口）。 url = self._snap_url("chv-a") self._record("chv-a", url, None) ex = self.make_exec(env_extra={ "CHV_WARM_POOL_SIZE": "0", "CHV_SNAPSHOT_STRICT_TENANT": "0"}) snapshot_url, _net = ex._validate_restore( {"id": "task-a", "project_id": "tenant-a"}, url) self.assertEqual(snapshot_url, url) self.assertIn("vm_restore_tenant_unmarked", self._event_names()) self.assertNotIn("vm_restore_denied", self._event_names()) # -- dispatch 入口接线（集成时补）-------------------------------------- class DispatchWiringTest(HybridACase): def test_dispatch_task_routes_to_hybrid_when_pool_on(self): # warm_pool_size>=1 时 dispatch_task 走混合路由。 ex = self.make_exec() out = ex.dispatch_task(self._task(task_id="t1")) self.assertEqual(out["status"], "completed") self.assertEqual(out["route"], "pool") self.assertEqual(out["allocation"], "pool-cold") def test_dispatch_task_routes_resident_when_paid(self): ex = self.make_exec() out = ex.dispatch_task( self._task(task_id="t1", project="paid1", tier="paid")) self.assertEqual(out["status"], "completed") self.assertEqual(out["route"], "resident") self.assertEqual(out["allocation"], "resident-cold") def test_dispatch_task_stays_cold_when_pool_off(self): # warm_pool_size==0 时 dispatch_task 保持旧冷启语义， # 不进混合路由（结果无 route 字段）。 ex = self.make_exec(env_extra={"CHV_WARM_POOL_SIZE": "0"}) out = ex.dispatch_task(self._task(task_id="t1")) self.assertEqual(out["status"], "completed") self.assertNotIn("route", out) if __name__ == "__main__": unittest.main()