refactor(vms): extract operations/vms.py seam from commands/vm.py (W1)
This commit is contained in:
@@ -378,110 +378,6 @@ def test_vm_remove_clone_dir_removal_failure(
|
||||
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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user