"""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 ( DATA_MOUNT_POLL_INTERVAL_SECONDS, DATA_MOUNT_TIMEOUT_SECONDS, 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 ( DATA_MOUNT_POLL_INTERVAL_SECONDS, DATA_MOUNT_TIMEOUT_SECONDS, 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 v54 已移除盘格式自动探测：热插盘 body 须与 create # 时盘同款带 image_type（租户数据盘是 raw 稀疏盘 → Raw）。 self.assertEqual(payload["image_type"], "Raw") # p2 第三轮：attach 不再以 chv 调 chv-data-mount（那条 # 路恒因权限失败、脚本又静默 exit 0），改为轮询等 # root hotplug 路径把盘挂上：ssh 里须出现 mountpoint # 轮询，且绝不能出现 chv-data-mount 调用。 self.assertTrue(any( "mountpoint -q /home/chv/workspace" in " ".join(call) for call in self.calls if call and call[0] == "ssh")) self.assertFalse(any( "/usr/local/bin/chv-data-mount" in " ".join(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_attach_failure_records_chv_body(self): # p2 修复：非 2xx 时 CHV 的原话（body）须进异常 detail # 与审计事件，不再丢弃；失败仍 fail-closed（不挂载、不 # 写归属标记）。 def api(socket_path, method, path, payload=None): self.api_calls.append((method, path, payload)) if path == "/vm.add-disk": return (400, {"error": "Image type required for disk"}) return (200, {}) ex = self.make_exec(api_client=api) with self.assertRaises(ChvFailure) as ctx: ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0") self.assertEqual(ctx.exception.kind, "data_disk_attach_failed") self.assertIn("Image type required for disk", str(ctx.exception)) self.assertIn("status=400", str(ctx.exception)) details = self._event_details("data_disk_attach_failed") self.assertEqual(len(details), 1) self.assertIn("Image type required for disk", details[0]) self.assertIn("status=400", details[0]) # 挂载轮询没跑（add-disk 已失败）、归属标记没写。 self.assertFalse(any( "mountpoint -q /home/chv/workspace" in " ".join(call) for call in self.calls if call and call[0] == "ssh")) self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0")) def test_attach_failure_truncates_long_body(self): # 超 500 字符的 body 只记前 500 字符并标明，不把整段 # 塞进审计与异常。 def api(socket_path, method, path, payload=None): self.api_calls.append((method, path, payload)) if path == "/vm.add-disk": return (400, {"error": "E" * 600}) return (200, {}) ex = self.make_exec(api_client=api) with self.assertRaises(ChvFailure) as ctx: ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0") self.assertIn("…(已截断)", str(ctx.exception)) self.assertNotIn("E" * 600, str(ctx.exception)) details = self._event_details("data_disk_attach_failed") self.assertEqual(len(details), 1) self.assertIn("…(已截断)", details[0]) 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) # -- p2 第三轮：挂盘等待 + 卸载提权（2026-11-09 第 2 轮复测）--------- class DataMountWaitAndPrivilegedUmountTest(HybridACase): """覆盖第 2 轮复测的两个根因： - 挂盘时序竞态：attach 须等客内 workspace 真挂上才返回 成功，超时 loud 失败；且绝不再以 chv 调 chv-data-mount 并判 rc==0（静默 exit 0 的成功假象）。 - 卸载权限：detach 须走 sudo -n 窄规则 umount，plain umount 恒 rc=32 的老路不得再出现；提权失败即 False， release 留置 VM 不回收。 只 mock 成功/失败路径的 SSH 返回码，不启动真实 VM。""" def _poll_calls(self): return [call for call in self.calls if call and call[0] == "ssh" and "mountpoint -q /home/chv/workspace" in " ".join(call)] def _mount_runner(self, succeed_after): """轮询专用 runner：前 succeed_after-1 次轮询报 未挂上（rc=1，第 2 轮真机：attach 返回时盘尚未挂）， 第 succeed_after 次起报已挂上；非轮询 ssh 一律 rc=0。 succeed_after=None 即永远未挂上。""" state = {"n": 0} def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd and cmd[0] == "ssh" and any( "mountpoint -q" in str(tok) for tok in cmd): state["n"] += 1 if succeed_after is not None \ and state["n"] >= succeed_after: return FakeResult(0, "", "") return FakeResult(1, "", "") return FakeResult(0, "", "") return runner def test_mount_poll_defaults(self): self.assertEqual(DATA_MOUNT_TIMEOUT_SECONDS, 60) self.assertEqual(DATA_MOUNT_POLL_INTERVAL_SECONDS, 2) def test_attach_waits_for_mount_before_returning(self): # 真机时序复现：add-disk 成功后前两次轮询盘未挂， # 第三次才挂上——attach 必须等到第三次才返回成功， # 而不是 add-disk 一成功就返回。 ex = self.make_exec( runner=self._mount_runner(succeed_after=3), env_extra={"CHV_DATA_MOUNT_TIMEOUT_SECONDS": "10", "CHV_DATA_MOUNT_INTERVAL_SECONDS": "1"}) out = ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0") self.assertEqual(out["tenant_id"], "tenant-a") self.assertEqual(len(self._poll_calls()), 3) self.assertEqual( read_vm_tenant(self.state_dir, "chv-pool-0"), "tenant-a") def test_attach_mount_timeout_fails_loud(self): # 盘一直不挂：超时后抛 data_disk_mount_timeout（kind # 同名），记同名审计事件，不写归属标记，不谎报成功。 ex = self.make_exec( runner=self._mount_runner(succeed_after=None), env_extra={"CHV_DATA_MOUNT_TIMEOUT_SECONDS": "3", "CHV_DATA_MOUNT_INTERVAL_SECONDS": "1"}) with self.assertRaises(ChvFailure) as ctx: ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0") self.assertEqual(ctx.exception.kind, "data_disk_mount_timeout") self.assertIn("3.0", str(ctx.exception)) self.assertEqual(len(self._poll_calls()), 3) details = self._event_details("data_disk_mount_timeout") self.assertEqual(len(details), 1) self.assertIn("chv-pool-0", details[0]) self.assertIsNone(read_vm_tenant(self.state_dir, "chv-pool-0")) def test_attach_never_runs_data_mount_as_chv(self): # 回归：静默 exit 0 的成功假象不得再出现——attach 全程 # 不得以 chv 身份调 /usr/local/bin/chv-data-mount。 ex = self.make_exec() ex.attach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0") self.assertFalse(any( "/usr/local/bin/chv-data-mount" in " ".join(call) for call in self.calls if call and call[0] == "ssh")) def test_detach_uses_privileged_umount(self): ex = self.make_exec() self.assertTrue(ex.detach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0")) sudo_calls = [call for call in self.calls if call and call[0] == "ssh" and "sudo" in call] self.assertEqual(len(sudo_calls), 1) call = sudo_calls[0] # 精确 argv：sudo -n /usr/bin/umount /home/chv/workspace， # 且同一调用里不得出现不带 sudo 的 plain umount。 self.assertIn("sudo", call) self.assertIn("-n", call) self.assertIn("/usr/bin/umount", call) self.assertIn("/home/chv/workspace", call) self.assertLess(call.index("sudo"), call.index("/usr/bin/umount")) self.assertFalse(any( call and call[0] == "ssh" and "sudo" not in call and "umount" in call for call in self.calls)) def test_detach_privilege_failure_returns_false(self): # sudo 规则缺失/镜像未重建（sudo 报 rc=1）即 False， # 不谎报卸载成功。 def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd and cmd[0] == "ssh" and "sudo" in cmd: return FakeResult(1, "", "sudo: not found") return FakeResult(0, "", "") ex = self.make_exec(runner=runner) self.assertFalse(ex.detach_tenant_data_disk( self._task(project="tenant-a"), "chv-pool-0")) def test_release_retains_vm_when_unmount_fails(self): # 第 2 轮根因 2 的终局：卸载失败 → release 返回 # released=False + reason=unmount_failed，记审计， # VM 不销毁（不丢改动后静默回收）。 def runner(cmd, timeout=None): self.calls.append(list(cmd)) if cmd and cmd[0] == "ssh" and "sudo" in cmd: return FakeResult(32, "", "umount: must be superuser") return FakeResult(0, "", "") ex = self.make_exec(runner=runner) out = ex.release_pool_vm( self._task(project="tenant-a"), "chv-pool-0") self.assertFalse(out["released"]) self.assertEqual(out["reason"], "unmount_failed") self.assertIn("warm_pool_release_failed", self._event_names()) details = self._event_details("warm_pool_release_failed") self.assertTrue(any("unmount_failed" in d for d in details)) class DataMountGuestWiringTest(unittest.TestCase): """客机侧接线（纯文本断言，不启动 VM）：chv-data-mount 失败 loud、build.sh 装 sudo + 写窄规则。""" def setUp(self): self.root = Path(__file__).resolve().parents[2] def test_data_mount_failures_are_loud(self): text = (self.root / "chv-guest" / "chv-data-mount.sh").read_text() # 盘已找到后 mkfs/mount 失败必须 exit 1，不再静默 0。 self.assertIn('mkfs failed on $DEV"; exit 1', text) self.assertIn('mount failed"; exit 1', text) # 无盘仍是启动期 no-op：exit 0，不挡 boot。 self.assertIn('no data disk found, skip"; exit 0', text) def test_build_sh_installs_sudo_and_narrow_rule(self): text = (self.root / "chv-guest" / "build.sh").read_text() self.assertIn("iptables sudo", text) self.assertIn( "chv ALL=(root) NOPASSWD: " "/usr/bin/umount /home/chv/workspace", text) self.assertIn("/etc/sudoers.d/chv-data-umount", text) self.assertIn("chmod 0440 /etc/sudoers.d/chv-data-umount", text) self.assertIn("visudo -cf /etc/sudoers.d/chv-data-umount", text) # 窄规则只放行 umount 这一条：不得出现放宽到 mount、 # ALL 命令或带 shell 的写法。 idx = text.find("/etc/sudoers.d/chv-data-umount") block = text[max(0, idx - 1500):idx + 500] self.assertNotIn("NOPASSWD: ALL", block) self.assertNotIn("/usr/bin/mount", block) def test_buildinfo_lists_sudo_and_umount_rule(self): text = (self.root / "chv-guest" / "buildinfo.py").read_text() self.assertIn('"sudo", "chv-data-umount"', text) if __name__ == "__main__": unittest.main()