refactor(ops): add operations/cache.py + operations/host.py seam (W2)
Extracts pure, click-free cargo-cache status/purge and host disk/runner/ backup-template logic into operations/, mirroring the W1 pattern. Routes commands/retention.py::_purge_cargo_cache through operations.cache while preserving identical CLI output. commands/monitor.py is left untouched (its refactor is optional per the plan; the Windows/Linux polymorphic _service_status and the disk not-found detection don't map cleanly onto the new pure functions without risking behavioural drift).
This commit is contained in:
@@ -0,0 +1,282 @@
|
||||
"""Tests for the pure ``operations.host`` layer (W2 of the web-UI plan).
|
||||
|
||||
Covers ``disk_status``, ``runner_status`` and ``backup_template`` — the seam
|
||||
both the CLI (``monitor disk``/``monitor runner``/``template backup``) and
|
||||
the future web UI import. Zero ``click``/``print`` in the module under test;
|
||||
no real ``systemctl``/``tar``/``7z`` is ever invoked — everything is
|
||||
monkeypatched.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
import ci_orchestrator.operations.host as host_module
|
||||
from ci_orchestrator.config import BackendConfig, Config, Paths
|
||||
from ci_orchestrator.operations.host import (
|
||||
BackupResult,
|
||||
DiskStatus,
|
||||
RunnerStatus,
|
||||
backup_template,
|
||||
disk_status,
|
||||
runner_status,
|
||||
)
|
||||
|
||||
|
||||
def _make_config(templates_dir: Path) -> Config:
|
||||
return Config(
|
||||
paths=Paths(
|
||||
root=templates_dir.parent,
|
||||
templates=templates_dir,
|
||||
build_vms=templates_dir.parent / "build-vms",
|
||||
artifacts=templates_dir.parent / "artifacts",
|
||||
keys=templates_dir.parent / "keys",
|
||||
),
|
||||
backend=BackendConfig(),
|
||||
)
|
||||
|
||||
|
||||
def _fake_usage(free_gb: float, total_gb: float = 500.0) -> shutil._ntuple_diskusage:
|
||||
gb = 1024**3
|
||||
return shutil._ntuple_diskusage(
|
||||
int(total_gb * gb), int((total_gb - free_gb) * gb), int(free_gb * gb)
|
||||
)
|
||||
|
||||
|
||||
# ── disk_status ──────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_disk_status_normal(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "disk_usage", lambda _p: _fake_usage(free_gb=250, total_gb=500))
|
||||
|
||||
status = disk_status(tmp_path)
|
||||
|
||||
assert status.path == tmp_path
|
||||
assert status.total_gb == pytest.approx(500.0, rel=1e-3)
|
||||
assert status.free_gb == pytest.approx(250.0, rel=1e-3)
|
||||
assert status.used_pct == pytest.approx(50.0, rel=1e-2)
|
||||
|
||||
|
||||
def test_disk_status_oserror_degrades_to_zeros(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
def _raise(_p: Any) -> Any:
|
||||
raise OSError("no such drive")
|
||||
|
||||
monkeypatch.setattr(host_module.shutil, "disk_usage", _raise)
|
||||
|
||||
status = disk_status(tmp_path)
|
||||
|
||||
assert status == DiskStatus(path=tmp_path, total_gb=0.0, free_gb=0.0, used_pct=0.0)
|
||||
|
||||
|
||||
# ── runner_status ────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class _RunResult:
|
||||
def __init__(self, stdout: str = "", returncode: int = 0) -> None:
|
||||
self.stdout = stdout
|
||||
self.returncode = returncode
|
||||
|
||||
|
||||
def test_runner_status_active(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
||||
|
||||
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
||||
if cmd[1] == "is-active":
|
||||
return _RunResult(stdout="active\n")
|
||||
return _RunResult(stdout="running\n")
|
||||
|
||||
monkeypatch.setattr(host_module.subprocess, "run", _run)
|
||||
|
||||
result = runner_status(["act-runner.service"])
|
||||
|
||||
assert result == [RunnerStatus(name="act-runner.service", active=True, sub_state="running")]
|
||||
|
||||
|
||||
def test_runner_status_inactive(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
||||
|
||||
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
||||
if cmd[1] == "is-active":
|
||||
return _RunResult(stdout="inactive\n", returncode=3)
|
||||
return _RunResult(stdout="dead\n")
|
||||
|
||||
monkeypatch.setattr(host_module.subprocess, "run", _run)
|
||||
|
||||
result = runner_status(["act-runner-rust.service"])
|
||||
|
||||
assert result == [RunnerStatus(name="act-runner-rust.service", active=False, sub_state="dead")]
|
||||
|
||||
|
||||
def test_runner_status_multiple_services(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
||||
|
||||
def _run(cmd: list[str], **_kw: Any) -> _RunResult:
|
||||
name = cmd[-1]
|
||||
if cmd[1] == "is-active":
|
||||
return _RunResult(stdout="active\n" if name == "a" else "inactive\n")
|
||||
return _RunResult(stdout="running\n" if name == "a" else "dead\n")
|
||||
|
||||
monkeypatch.setattr(host_module.subprocess, "run", _run)
|
||||
|
||||
result = runner_status(["a", "b"])
|
||||
|
||||
assert result == [
|
||||
RunnerStatus(name="a", active=True, sub_state="running"),
|
||||
RunnerStatus(name="b", active=False, sub_state="dead"),
|
||||
]
|
||||
|
||||
|
||||
def test_runner_status_systemctl_absent(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: None)
|
||||
|
||||
result = runner_status(["act-runner.service", "act-runner-rust.service"])
|
||||
|
||||
assert result == [
|
||||
RunnerStatus(name="act-runner.service", active=False, sub_state="unknown"),
|
||||
RunnerStatus(name="act-runner-rust.service", active=False, sub_state="unknown"),
|
||||
]
|
||||
|
||||
|
||||
def test_runner_status_timeout_tolerated(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
||||
|
||||
def _boom(*_a: Any, **_kw: Any) -> Any:
|
||||
raise subprocess.TimeoutExpired(cmd="systemctl", timeout=10.0)
|
||||
|
||||
monkeypatch.setattr(host_module.subprocess, "run", _boom)
|
||||
|
||||
result = runner_status(["act-runner.service"])
|
||||
|
||||
assert result == [RunnerStatus(name="act-runner.service", active=False, sub_state="unknown")]
|
||||
|
||||
|
||||
def test_runner_status_oserror_tolerated(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
||||
|
||||
def _boom(*_a: Any, **_kw: Any) -> Any:
|
||||
raise OSError("boom")
|
||||
|
||||
monkeypatch.setattr(host_module.subprocess, "run", _boom)
|
||||
|
||||
result = runner_status(["svc"])
|
||||
|
||||
assert result == [RunnerStatus(name="svc", active=False, sub_state="unknown")]
|
||||
|
||||
|
||||
def test_runner_status_never_uses_shell_true(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setattr(host_module.shutil, "which", lambda _n: "/usr/bin/systemctl")
|
||||
seen_kwargs: list[dict[str, Any]] = []
|
||||
|
||||
def _run(cmd: list[str], **kw: Any) -> _RunResult:
|
||||
seen_kwargs.append(kw)
|
||||
return _RunResult(stdout="active\n")
|
||||
|
||||
monkeypatch.setattr(host_module.subprocess, "run", _run)
|
||||
runner_status(["svc"])
|
||||
|
||||
assert all(not kw.get("shell") for kw in seen_kwargs)
|
||||
|
||||
|
||||
# ── backup_template ─────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_backup_template_missing_template_dir_errors(tmp_path: Path) -> None:
|
||||
templates = tmp_path / "templates"
|
||||
templates.mkdir()
|
||||
config = _make_config(templates)
|
||||
|
||||
result = backup_template(config, "NoSuchTemplate")
|
||||
|
||||
assert result.ok is False
|
||||
assert result.archive_path is None
|
||||
assert result.error is not None
|
||||
assert "NoSuchTemplate" in result.error or "not found" in result.error.lower()
|
||||
|
||||
|
||||
def test_backup_template_dry_run_does_not_compress(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
templates = tmp_path / "templates"
|
||||
tmpl_dir = templates / "WinBuild2025"
|
||||
tmpl_dir.mkdir(parents=True)
|
||||
config = _make_config(templates)
|
||||
monkeypatch.setattr(host_module, "_BACKUP_BASE_DIR", tmp_path / "backups")
|
||||
|
||||
calls: list[tuple[Path, Path]] = []
|
||||
monkeypatch.setattr(host_module, "_compress_7z", lambda src, dst: calls.append((src, dst)))
|
||||
|
||||
result = backup_template(config, "WinBuild2025", dry_run=True)
|
||||
|
||||
assert result.ok is True
|
||||
assert result.template == "WinBuild2025"
|
||||
assert result.archive_path is not None
|
||||
assert result.error is None
|
||||
assert calls == [], "dry_run must not compress"
|
||||
assert not result.archive_path.exists()
|
||||
|
||||
|
||||
def test_backup_template_success_mocked_archive_step(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
templates = tmp_path / "templates"
|
||||
tmpl_dir = templates / "WinBuild2025"
|
||||
tmpl_dir.mkdir(parents=True)
|
||||
config = _make_config(templates)
|
||||
monkeypatch.setattr(host_module, "_BACKUP_BASE_DIR", tmp_path / "backups")
|
||||
|
||||
def _fake_compress(src: Path, dst: Path) -> None:
|
||||
dst.parent.mkdir(parents=True, exist_ok=True)
|
||||
dst.write_bytes(b"archive")
|
||||
|
||||
monkeypatch.setattr(host_module, "_compress_7z", _fake_compress)
|
||||
|
||||
result = backup_template(config, "WinBuild2025")
|
||||
|
||||
assert result == BackupResult(
|
||||
template="WinBuild2025", archive_path=result.archive_path, ok=True, error=None
|
||||
)
|
||||
assert result.archive_path is not None
|
||||
assert result.archive_path.exists()
|
||||
|
||||
|
||||
def test_backup_template_archive_failure_reported(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
templates = tmp_path / "templates"
|
||||
tmpl_dir = templates / "WinBuild2025"
|
||||
tmpl_dir.mkdir(parents=True)
|
||||
config = _make_config(templates)
|
||||
monkeypatch.setattr(host_module, "_BACKUP_BASE_DIR", tmp_path / "backups")
|
||||
|
||||
def _boom(_src: Path, _dst: Path) -> None:
|
||||
raise subprocess.CalledProcessError(1, ["7z"])
|
||||
|
||||
monkeypatch.setattr(host_module, "_compress_7z", _boom)
|
||||
|
||||
result = backup_template(config, "WinBuild2025")
|
||||
|
||||
assert result.ok is False
|
||||
assert result.archive_path is None
|
||||
assert result.error is not None
|
||||
|
||||
|
||||
def test_backup_template_never_calls_real_vmrun_or_7z(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
"""Guard: without mocking ``_compress_7z`` this would try to run real ``7z``."""
|
||||
templates = tmp_path / "templates"
|
||||
tmpl_dir = templates / "WinBuild2025"
|
||||
tmpl_dir.mkdir(parents=True)
|
||||
config = _make_config(templates)
|
||||
monkeypatch.setattr(host_module, "_BACKUP_BASE_DIR", tmp_path / "backups")
|
||||
|
||||
called: list[bool] = []
|
||||
|
||||
def _fake(_src: Path, _dst: Path) -> None:
|
||||
called.append(True)
|
||||
|
||||
monkeypatch.setattr(host_module, "_compress_7z", _fake)
|
||||
backup_template(config, "WinBuild2025")
|
||||
|
||||
assert called == [True]
|
||||
Reference in New Issue
Block a user