Simone
b2b31f4b6e
Bring monitor.py 49->99%%, vm.py 79->98%%, winrm.py 76->100%%. Total project coverage 80.10%% -> 90.69%%, satisfying the 90%% gate set in commit 451a61c.
500 lines
16 KiB
Python
500 lines
16 KiB
Python
"""Tests for ``ci_orchestrator vm remove`` and ``vm cleanup``.
|
|
|
|
Migrated from Pester ``tests/Remove-BuildVM.Tests.ps1`` (now removed).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
import ci_orchestrator.commands.vm as vm_module
|
|
from ci_orchestrator.__main__ import cli
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
|
|
|
|
class _FakeBackend:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
running_after_stop: bool = False,
|
|
delete_fails_first: int = 0,
|
|
) -> None:
|
|
self.calls: list[tuple[str, Any]] = []
|
|
self._running = True
|
|
self._running_after_stop = running_after_stop
|
|
self._delete_fails_remaining = delete_fails_first
|
|
|
|
def stop(self, handle: VmHandle, hard: bool = False) -> None:
|
|
self.calls.append(("stop", hard))
|
|
if not hard or not self._running_after_stop:
|
|
self._running = False
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
return self._running
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
self.calls.append(("delete", None))
|
|
if self._delete_fails_remaining > 0:
|
|
self._delete_fails_remaining -= 1
|
|
raise BackendOperationFailed("deleteVM", 1, "transient")
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _no_sleep(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(time, "sleep", lambda _s: None)
|
|
monkeypatch.setattr(vm_module.time, "sleep", lambda _s: None)
|
|
|
|
|
|
def _make_clone_dir(tmp_path: Path, name: str = "clone") -> tuple[Path, Path]:
|
|
clone_dir = tmp_path / name
|
|
clone_dir.mkdir()
|
|
vmx = clone_dir / f"{name}.vmx"
|
|
vmx.write_text('config.version = "8"', encoding="utf-8")
|
|
return clone_dir, vmx
|
|
|
|
|
|
# ── vm remove ──────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_vm_remove_missing_vmx_returns_without_error(tmp_path: Path) -> None:
|
|
"""Pester migration: returns without error when VMX does not exist."""
|
|
missing = tmp_path / "ghost" / "ghost.vmx"
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(missing)])
|
|
assert result.exit_code == 0, result.output
|
|
assert "nothing to remove" in result.output
|
|
|
|
|
|
def test_vm_remove_partial_clone_dir_is_removed(tmp_path: Path) -> None:
|
|
"""Pester migration: partial clone dir without VMX is still removed."""
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "partial")
|
|
vmx.unlink() # leave the directory but no VMX
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
|
|
assert result.exit_code == 0
|
|
assert not clone_dir.exists()
|
|
|
|
|
|
def test_vm_remove_happy_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
"""Pester migration: removes the clone dir when VMX exists and stop+delete succeed."""
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "live")
|
|
backend = _FakeBackend()
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
|
|
result = CliRunner().invoke(
|
|
cli, ["vm", "remove", "--vmx", str(vmx), "--graceful-timeout", "1"]
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert ("delete", None) in backend.calls
|
|
assert not clone_dir.exists()
|
|
|
|
|
|
def test_vm_remove_force_skips_soft_stop(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
_, vmx = _make_clone_dir(tmp_path, "force")
|
|
backend = _FakeBackend()
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx), "--force"])
|
|
assert result.exit_code == 0, result.output
|
|
# No soft stop was issued.
|
|
soft_stops = [c for c in backend.calls if c == ("stop", False)]
|
|
assert soft_stops == []
|
|
hard_stops = [c for c in backend.calls if c == ("stop", True)]
|
|
assert hard_stops != []
|
|
|
|
|
|
def test_vm_remove_delete_retry_succeeds(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "retry")
|
|
backend = _FakeBackend(delete_fails_first=2)
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
|
|
assert result.exit_code == 0
|
|
delete_calls = [c for c in backend.calls if c[0] == "delete"]
|
|
assert len(delete_calls) == 3 # 2 failures + 1 success
|
|
assert not clone_dir.exists()
|
|
|
|
|
|
# ── vm cleanup ─────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _age_dir(path: Path, hours: float) -> None:
|
|
ts = (datetime.now() - timedelta(hours=hours)).timestamp()
|
|
import os
|
|
|
|
os.utime(path, (ts, ts))
|
|
|
|
|
|
def test_vm_cleanup_dry_run_lists_orphans(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
base = tmp_path / "build-vms"
|
|
base.mkdir()
|
|
fresh, _ = _make_clone_dir(base, "fresh")
|
|
old, _ = _make_clone_dir(base, "old")
|
|
_age_dir(old, hours=10)
|
|
_age_dir(fresh, hours=0)
|
|
# Make sure backend isn't constructed (vmrun absent on test host).
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: None)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"vm",
|
|
"cleanup",
|
|
"--clone-base-dir",
|
|
str(base),
|
|
"--max-age-hours",
|
|
"4",
|
|
"--what-if",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert "would destroy" in result.output
|
|
assert old.exists() # dry run did NOT delete
|
|
assert fresh.exists()
|
|
|
|
|
|
def test_vm_cleanup_destroys_old_dirs(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
base = tmp_path / "build-vms"
|
|
base.mkdir()
|
|
old, _ = _make_clone_dir(base, "old")
|
|
_age_dir(old, hours=10)
|
|
|
|
backend = _FakeBackend()
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
|
|
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"vm",
|
|
"cleanup",
|
|
"--clone-base-dir",
|
|
str(base),
|
|
"--max-age-hours",
|
|
"4",
|
|
],
|
|
)
|
|
assert result.exit_code == 0, result.output
|
|
assert not old.exists()
|
|
# Backend was used to stop+delete.
|
|
assert any(c[0] == "delete" for c in backend.calls)
|
|
|
|
|
|
def test_vm_cleanup_handles_missing_base_dir(tmp_path: Path) -> None:
|
|
missing = tmp_path / "does-not-exist"
|
|
result = CliRunner().invoke(
|
|
cli, ["vm", "cleanup", "--clone-base-dir", str(missing)]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "nothing to do" in result.output
|
|
|
|
|
|
def test_vm_cleanup_removes_stale_lock(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
base = tmp_path / "bv"
|
|
base.mkdir()
|
|
lock = tmp_path / "vm-start.lock"
|
|
lock.write_text("x")
|
|
_age_dir(lock, hours=1)
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: None)
|
|
result = CliRunner().invoke(
|
|
cli,
|
|
[
|
|
"vm",
|
|
"cleanup",
|
|
"--clone-base-dir",
|
|
str(base),
|
|
"--lock-file",
|
|
str(lock),
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
assert not lock.exists()
|
|
|
|
|
|
# ── helpers + uncovered branches ──────────────────────────────────────────
|
|
|
|
|
|
def test_make_backend_with_vmrun_path(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
"""Override path returns a WorkstationVmrunBackend honouring vmrun_path."""
|
|
from ci_orchestrator.backends.workstation import WorkstationVmrunBackend
|
|
|
|
class _StubConfig:
|
|
pass
|
|
|
|
fake_vmrun = tmp_path / "vmrun.exe"
|
|
fake_vmrun.write_text("x")
|
|
monkeypatch.setattr(vm_module, "load_config", lambda: _StubConfig())
|
|
backend = vm_module._make_backend(str(fake_vmrun))
|
|
assert isinstance(backend, WorkstationVmrunBackend)
|
|
|
|
|
|
def test_make_backend_default_uses_load_backend(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
sentinel = object()
|
|
monkeypatch.setattr(vm_module, "load_config", lambda: "cfg")
|
|
monkeypatch.setattr(vm_module, "load_backend", lambda _c: sentinel)
|
|
assert vm_module._make_backend(None) is sentinel
|
|
|
|
|
|
def test_try_remove_dir_returns_true_when_missing(tmp_path: Path) -> None:
|
|
assert vm_module._try_remove_dir(tmp_path / "ghost") is True
|
|
|
|
|
|
def test_try_remove_dir_real_removal(tmp_path: Path) -> None:
|
|
d = tmp_path / "x"
|
|
(d / "sub").mkdir(parents=True)
|
|
(d / "sub" / "f.txt").write_text("a")
|
|
assert vm_module._try_remove_dir(d) is True
|
|
assert not d.exists()
|
|
|
|
|
|
def test_stop_with_fallback_soft_error_then_running_with_hard_error() -> None:
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
|
|
class _B:
|
|
def __init__(self) -> None:
|
|
self.calls: list[tuple[str, bool]] = []
|
|
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
self.calls.append(("stop", hard))
|
|
raise BackendOperationFailed("stop", 1, "boom")
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
return True
|
|
|
|
b = _B()
|
|
# Should swallow both errors.
|
|
vm_module._stop_with_fallback(b, VmHandle(identifier="x.vmx"))
|
|
assert ("stop", False) in b.calls
|
|
assert ("stop", True) in b.calls
|
|
|
|
|
|
def test_vm_remove_backend_unavailable(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
from ci_orchestrator.backends.errors import BackendNotAvailable
|
|
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "noback")
|
|
|
|
def _raise(_v: Any) -> Any:
|
|
raise BackendNotAvailable("vmrun missing")
|
|
|
|
monkeypatch.setattr(vm_module, "_make_backend", _raise)
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
|
|
assert result.exit_code == 0, result.output
|
|
assert "vmrun unavailable" in result.output
|
|
assert not clone_dir.exists()
|
|
|
|
|
|
def test_vm_remove_force_hard_stop_error(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
|
|
_, vmx = _make_clone_dir(tmp_path, "forcefail")
|
|
|
|
class _B:
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
raise BackendOperationFailed("stop", 1, "no")
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
return False
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
pass
|
|
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: _B())
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx), "--force"])
|
|
assert result.exit_code == 0
|
|
assert "hard stop failed" in result.output
|
|
|
|
|
|
def test_vm_remove_deadline_loop_with_running_then_error(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""Cover the deadline loop that breaks on BackendError."""
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
|
|
_, vmx = _make_clone_dir(tmp_path, "loop")
|
|
|
|
class _B:
|
|
def __init__(self) -> None:
|
|
self._calls = 0
|
|
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
pass
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
self._calls += 1
|
|
if self._calls == 1:
|
|
return True
|
|
raise BackendOperationFailed("isrunning", 1, "lost")
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
pass
|
|
|
|
# Make monotonic strictly advance so the loop runs at least twice but exits.
|
|
base_t = [0.0]
|
|
|
|
def _mono() -> float:
|
|
base_t[0] += 1.0
|
|
return base_t[0]
|
|
|
|
monkeypatch.setattr(vm_module.time, "monotonic", _mono)
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: _B())
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
|
|
assert result.exit_code == 0
|
|
|
|
|
|
def test_vm_remove_delete_fails_all_retries(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""Delete that fails on every retry triggers fallback dir-removal log."""
|
|
clone_dir, vmx = _make_clone_dir(tmp_path, "neverdeletes")
|
|
|
|
class _B:
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
pass
|
|
|
|
def is_running(self, _h: VmHandle) -> bool:
|
|
return False
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
raise BackendOperationFailed("delete", 1, "always")
|
|
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: _B())
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
|
|
assert result.exit_code == 0
|
|
assert "deleteVM failed after retries" in result.output
|
|
assert not clone_dir.exists()
|
|
|
|
|
|
def test_vm_remove_clone_dir_removal_failure(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
"""When _try_remove_dir reports failure, the manual-cleanup notice fires."""
|
|
_, vmx = _make_clone_dir(tmp_path, "stubborn")
|
|
backend = _FakeBackend()
|
|
monkeypatch.setattr(vm_module, "_make_backend", lambda _v: backend)
|
|
monkeypatch.setattr(vm_module, "_try_remove_dir", lambda _p: False)
|
|
result = CliRunner().invoke(cli, ["vm", "remove", "--vmx", str(vmx)])
|
|
assert result.exit_code == 0
|
|
assert "manual cleanup needed" in result.output
|
|
|
|
|
|
def test_list_orphans_non_dir_base(tmp_path: Path) -> None:
|
|
f = tmp_path / "file.txt"
|
|
f.write_text("x")
|
|
assert vm_module._list_orphans(f, 4) == []
|
|
|
|
|
|
def test_list_orphans_skips_non_dir_and_oserror(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
(base / "a-file").write_text("not a dir")
|
|
bad_dir = base / "bad"
|
|
bad_dir.mkdir()
|
|
_age_dir(bad_dir, hours=10)
|
|
|
|
real_stat = Path.stat
|
|
|
|
def _stat(self: Path, *a: Any, **kw: Any) -> Any:
|
|
if self == bad_dir:
|
|
raise OSError("nope")
|
|
return real_stat(self, *a, **kw)
|
|
|
|
monkeypatch.setattr(Path, "stat", _stat)
|
|
assert vm_module._list_orphans(base, 4) == []
|
|
|
|
|
|
def test_find_vmx_returns_none(tmp_path: Path) -> None:
|
|
d = tmp_path / "novmx"
|
|
d.mkdir()
|
|
(d / "readme.txt").write_text("x")
|
|
assert vm_module._find_vmx(d) is None
|
|
|
|
|
|
def test_cleanup_orphans_dir_without_vmx_logs_and_removes(tmp_path: Path) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
no_vmx = base / "no-vmx"
|
|
no_vmx.mkdir()
|
|
_age_dir(no_vmx, hours=10)
|
|
runner_count = vm_module.cleanup_orphans(
|
|
clone_base=base, max_age_hours=4, backend=None
|
|
)
|
|
assert runner_count == 1
|
|
assert not no_vmx.exists()
|
|
|
|
|
|
def test_cleanup_orphans_backend_stop_error_swallowed(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
from ci_orchestrator.backends.errors import BackendOperationFailed
|
|
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
old, _ = _make_clone_dir(base, "old1")
|
|
_age_dir(old, hours=10)
|
|
|
|
class _B:
|
|
def stop(self, _h: VmHandle, hard: bool = False) -> None:
|
|
raise BackendOperationFailed("stop", 1, "no")
|
|
|
|
def delete(self, _h: VmHandle) -> None:
|
|
raise BackendOperationFailed("delete", 1, "no")
|
|
|
|
count = vm_module.cleanup_orphans(
|
|
clone_base=base, max_age_hours=4, backend=_B()
|
|
)
|
|
assert count == 1
|
|
|
|
|
|
def test_cleanup_orphans_dir_removal_failure_logs(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
old, _ = _make_clone_dir(base, "stuck")
|
|
_age_dir(old, hours=10)
|
|
monkeypatch.setattr(vm_module, "_try_remove_dir", lambda _p: False)
|
|
count = vm_module.cleanup_orphans(
|
|
clone_base=base, max_age_hours=4, backend=None
|
|
)
|
|
assert count == 0
|
|
|
|
|
|
def test_cleanup_orphans_lock_unlink_oserror(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
base = tmp_path / "b"
|
|
base.mkdir()
|
|
lock = tmp_path / "stale.lock"
|
|
lock.write_text("x")
|
|
_age_dir(lock, hours=1)
|
|
|
|
real_unlink = Path.unlink
|
|
|
|
def _boom(self: Path, *a: Any, **kw: Any) -> Any:
|
|
if self == lock:
|
|
raise OSError("locked")
|
|
return real_unlink(self, *a, **kw)
|
|
|
|
monkeypatch.setattr(Path, "unlink", _boom)
|
|
vm_module.cleanup_orphans(
|
|
clone_base=base, max_age_hours=4, backend=None, lock_file=lock
|
|
)
|
|
|
|
|
|
def test_vm_cleanup_backend_unavailable(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
|
from ci_orchestrator.backends.errors import BackendNotAvailable
|
|
|
|
base = tmp_path / "build-vms"
|
|
base.mkdir()
|
|
|
|
def _raise(_v: Any) -> Any:
|
|
raise BackendNotAvailable("no vmrun")
|
|
|
|
monkeypatch.setattr(vm_module, "_make_backend", _raise)
|
|
result = CliRunner().invoke(
|
|
cli, ["vm", "cleanup", "--clone-base-dir", str(base)]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "vmrun not available" in result.output
|