diff --git a/src/ci_orchestrator/commands/retention.py b/src/ci_orchestrator/commands/retention.py index cb4748a..9dc37fa 100644 --- a/src/ci_orchestrator/commands/retention.py +++ b/src/ci_orchestrator/commands/retention.py @@ -14,8 +14,8 @@ from pathlib import Path import click -from ci_orchestrator.cargo_cache import CargoCacheStore from ci_orchestrator.config import Config, load_config +from ci_orchestrator.operations import cache as ops_cache @click.group() @@ -137,7 +137,7 @@ def _purge_old_dirs(base: Path, label: str, cutoff: datetime, *, what_if: bool) def _purge_cargo_cache(config: Config, effective_days: int, *, what_if: bool) -> None: - """Purge stale cargo cache entries. + """Purge stale cargo cache entries via :mod:`operations.cache`. Entries are the heaviest thing this policy governs -- hundreds of megabytes each, one per Cargo.lock change -- so leaving them out would repeat TODO @@ -153,25 +153,19 @@ def _purge_cargo_cache(config: Config, effective_days: int, *, what_if: bool) -> days = min(effective_days, config.cargo_cache.retention_days) cutoff = datetime.now(tz=UTC) - timedelta(days=days) - store = CargoCacheStore(config.cargo_cache.root) - # Size the victims first: after a real prune the files are gone and their - # bytes are unreadable, which is how a policy ends up reporting 0.0 GB - # freed while doing real work (TODO 2.9). - doomed = store.prune_older_than(cutoff, what_if=True) - if not doomed: + result = ops_cache.purge_cargo_cache(config, retention_days=days, dry_run=what_if) + + if result.removed_entries == 0: click.echo( f"[RetentionPolicy] cargo cache: nothing to purge " f"(cutoff: {cutoff.strftime('%Y-%m-%d')})." ) return - freed_mb = sum(p.stat().st_size for p in doomed if p.exists()) / 1_048_576 - - if not what_if: - store.prune_older_than(cutoff) + freed_mb = result.freed_bytes / 1_048_576 verb = "WhatIf: would purge" if what_if else "Purged" click.echo( - f"[RetentionPolicy] {verb} {len(doomed)} cargo cache " - f"{'entry' if len(doomed) == 1 else 'entries'} ({freed_mb:.0f} MB)" + f"[RetentionPolicy] {verb} {result.removed_entries} cargo cache " + f"{'entry' if result.removed_entries == 1 else 'entries'} ({freed_mb:.0f} MB)" ) diff --git a/src/ci_orchestrator/operations/cache.py b/src/ci_orchestrator/operations/cache.py new file mode 100644 index 0000000..a130479 --- /dev/null +++ b/src/ci_orchestrator/operations/cache.py @@ -0,0 +1,121 @@ +"""Pure cargo-cache operations shared by the CLI and (future) web UI. + +Wraps :class:`~ci_orchestrator.cargo_cache.CargoCacheStore` for status and +purge. Zero ``click``, zero ``print`` — everything here returns a dataclass. +Presentation stays in the CLI command wrappers (``commands/retention.py``) +and, later, the web routes. + +Scope note: this module wraps only the cargo cache's own status/purge. It is +deliberately orthogonal to the disk-retention *policy* (artifact/log purge, +aggressive-mode threshold) which lives in ``commands/retention.py`` — see +TODO §2.9 for that separate, still-open gap. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import TYPE_CHECKING + +from ci_orchestrator.cargo_cache import CargoCacheStore + +if TYPE_CHECKING: # pragma: no cover + from ci_orchestrator.config import Config + + +@dataclass(frozen=True) +class CacheStatus: + """Point-in-time snapshot of the cargo cache store.""" + + root: Path + total_bytes: int + entry_count: int + oldest_age_days: int | None + retention_days: int + + +@dataclass(frozen=True) +class PurgeResult: + """Outcome of a (real or dry-run) cargo cache purge.""" + + removed_entries: int + freed_bytes: int + dry_run: bool + + +def cargo_cache_status(config: Config) -> CacheStatus: + """Scan the cargo cache store and report size/age. Never raises. + + Reports the store's real on-disk state regardless of + ``config.cargo_cache.enabled`` — only :func:`purge_cargo_cache` respects + that flag. An absent store root reports an empty, zeroed status. + """ + root = config.cargo_cache.root + retention_days = config.cargo_cache.retention_days + + if not root.is_dir(): + return CacheStatus( + root=root, + total_bytes=0, + entry_count=0, + oldest_age_days=None, + retention_days=retention_days, + ) + + store = CargoCacheStore(root) + now = datetime.now(tz=UTC) + total_bytes = 0 + entry_count = 0 + oldest_mtime: datetime | None = None + + for entry in sorted(store.root.glob(f"*{store.suffix}")): + if not entry.is_file(): + continue + try: + st = entry.stat() + except OSError: + continue + total_bytes += st.st_size + entry_count += 1 + mtime = datetime.fromtimestamp(st.st_mtime, tz=UTC) + if oldest_mtime is None or mtime < oldest_mtime: + oldest_mtime = mtime + + oldest_age_days = (now - oldest_mtime).days if oldest_mtime is not None else None + + return CacheStatus( + root=root, + total_bytes=total_bytes, + entry_count=entry_count, + oldest_age_days=oldest_age_days, + retention_days=retention_days, + ) + + +def purge_cargo_cache( + config: Config, *, retention_days: int | None = None, dry_run: bool = False +) -> PurgeResult: + """Remove cache entries older than ``retention_days`` (or config's). + + Sizes the victims before deleting (a real prune makes their bytes + unreadable), so ``freed_bytes`` is accurate even for a real purge. + A disabled cache is left untouched and reports an all-zero result. + """ + if not config.cargo_cache.enabled: + return PurgeResult(removed_entries=0, freed_bytes=0, dry_run=dry_run) + + days = retention_days if retention_days is not None else config.cargo_cache.retention_days + cutoff = datetime.now(tz=UTC) - timedelta(days=days) + store = CargoCacheStore(config.cargo_cache.root) + + doomed = store.prune_older_than(cutoff, what_if=True) + freed_bytes = sum(p.stat().st_size for p in doomed if p.exists()) + + if not dry_run: + store.prune_older_than(cutoff) + + return PurgeResult(removed_entries=len(doomed), freed_bytes=freed_bytes, dry_run=dry_run) + + +__all__ = ["CacheStatus", "PurgeResult", "cargo_cache_status", "purge_cargo_cache"] diff --git a/src/ci_orchestrator/operations/host.py b/src/ci_orchestrator/operations/host.py new file mode 100644 index 0000000..c2c495d --- /dev/null +++ b/src/ci_orchestrator/operations/host.py @@ -0,0 +1,166 @@ +"""Pure host operations shared by the CLI and (future) web UI. + +Disk usage, runner (systemd unit) status, and template backup. Zero +``click``, zero ``print`` — everything here returns a dataclass. +Presentation stays in the CLI command wrappers and, later, the web routes. + +``runner_status`` is Linux/systemd-only (the Linux host runs no other +service manager per Phase C) and never shells out with ``shell=True`` — +every ``systemctl`` invocation is an argument list, read-only. + +``backup_template`` reuses the archive-step shape from +``commands/template.py::template_backup`` (timestamped ``7z`` archive under +``/var/lib/ci/backups``) but is a fresh, click-free implementation so this +module has no dependency on the CLI layer. +""" + +from __future__ import annotations + +import shutil +import subprocess +from dataclasses import dataclass +from datetime import UTC, datetime +from pathlib import Path +from typing import TYPE_CHECKING + +if TYPE_CHECKING: # pragma: no cover + from ci_orchestrator.config import Config + +_SYSTEMCTL_TIMEOUT = 10.0 +_BACKUP_BASE_DIR = Path("/var/lib/ci/backups") + + +@dataclass(frozen=True) +class DiskStatus: + """Filesystem usage for a single path/mount.""" + + path: Path + total_gb: float + free_gb: float + used_pct: float + + +@dataclass(frozen=True) +class RunnerStatus: + """``systemctl`` status for a single service.""" + + name: str + active: bool + sub_state: str + + +@dataclass(frozen=True) +class BackupResult: + """Outcome of a (real or dry-run) template backup.""" + + template: str + archive_path: Path | None + ok: bool + error: str | None + + +def disk_status(path: Path) -> DiskStatus: + """Report usage for ``path``. Degrades to zeros on ``OSError`` — never raises.""" + try: + usage = shutil.disk_usage(path) + except OSError: + return DiskStatus(path=path, total_gb=0.0, free_gb=0.0, used_pct=0.0) + + gb = 1024.0**3 + total_gb = usage.total / gb + free_gb = usage.free / gb + used_pct = ((usage.total - usage.free) / usage.total * 100.0) if usage.total > 0 else 0.0 + + return DiskStatus(path=path, total_gb=total_gb, free_gb=free_gb, used_pct=used_pct) + + +def _systemctl_query(systemctl: str, args: list[str]) -> str | None: + """Run a read-only ``systemctl`` query. Returns stripped stdout, or ``None`` on failure.""" + try: + result = subprocess.run( + [systemctl, *args], + capture_output=True, + text=True, + check=False, + timeout=_SYSTEMCTL_TIMEOUT, + ) + except (subprocess.TimeoutExpired, OSError): + return None + return result.stdout.strip() + + +def runner_status(service_names: list[str]) -> list[RunnerStatus]: + """Query ``systemctl is-active`` / ``show -p SubState`` for each service. + + Never raises: a missing ``systemctl``, a timeout, or an ``OSError`` all + degrade a service to ``active=False, sub_state="unknown"`` rather than + aborting the whole batch. + """ + systemctl = shutil.which("systemctl") + if systemctl is None: + return [RunnerStatus(name=name, active=False, sub_state="unknown") for name in service_names] + + out: list[RunnerStatus] = [] + for name in service_names: + active_raw = _systemctl_query(systemctl, ["is-active", name]) + sub_state_raw = _systemctl_query(systemctl, ["show", "-p", "SubState", "--value", name]) + active = active_raw == "active" + sub_state = sub_state_raw if sub_state_raw else "unknown" + out.append(RunnerStatus(name=name, active=active, sub_state=sub_state)) + return out + + +def _compress_7z(source: Path, archive: Path) -> None: + """Create a 7z archive of ``source`` at ``archive`` using compression level 1. + + Split out so tests can monkeypatch the actual compression step and never + invoke a real ``7z`` subprocess. + """ + subprocess.run( + ["7z", "a", "-mx=1", "-mmt=on", str(archive), str(source)], + check=True, + stdout=subprocess.DEVNULL, + stderr=subprocess.PIPE, + ) + + +def backup_template(config: Config, template: str, *, dry_run: bool = False) -> BackupResult: + """Archive a template directory to a timestamped ``7z`` file. + + ``dry_run`` computes and returns the would-be archive path without + compressing anything. On failure (missing template dir, or the + compression step raising), returns ``ok=False`` with ``archive_path=None`` + and never raises. + """ + tmpl_dir = config.paths.templates / template + if not tmpl_dir.is_dir(): + return BackupResult( + template=template, + archive_path=None, + ok=False, + error=f"Template directory not found: {tmpl_dir}", + ) + + timestamp = datetime.now(tz=UTC).strftime("%Y%m%d_%H%M%S") + archive_path = _BACKUP_BASE_DIR / f"Template_{template}_{timestamp}.7z" + + if dry_run: + return BackupResult(template=template, archive_path=archive_path, ok=True, error=None) + + try: + _BACKUP_BASE_DIR.mkdir(parents=True, exist_ok=True) + _compress_7z(tmpl_dir, archive_path) + except (OSError, subprocess.CalledProcessError) as exc: + return BackupResult(template=template, archive_path=None, ok=False, error=str(exc)) + + return BackupResult(template=template, archive_path=archive_path, ok=True, error=None) + + +__all__ = [ + "BackupResult", + "DiskStatus", + "RunnerStatus", + "backup_template", + "disk_status", + "runner_status", +] diff --git a/tests/python/test_operations_cache.py b/tests/python/test_operations_cache.py new file mode 100644 index 0000000..e989a0f --- /dev/null +++ b/tests/python/test_operations_cache.py @@ -0,0 +1,187 @@ +"""Tests for the pure ``operations.cache`` layer (W2 of the web-UI plan). + +Covers ``cargo_cache_status`` and ``purge_cargo_cache`` — the seam both the +CLI (``retention run``) and the future web UI import for cargo-cache +state/maintenance. Zero ``click``/``print`` in the module under test. +""" + +from __future__ import annotations + +import time +from pathlib import Path + +from ci_orchestrator.cargo_cache import CargoCacheStore, cache_key +from ci_orchestrator.config import BackendConfig, CargoCacheConfig, Config, Paths +from ci_orchestrator.operations.cache import ( + CacheStatus, + PurgeResult, + cargo_cache_status, + purge_cargo_cache, +) + + +def _make_config(root: Path, *, enabled: bool = True, retention_days: int = 30) -> Config: + return Config( + paths=Paths( + root=root.parent, + templates=root.parent / "templates", + build_vms=root.parent / "build-vms", + artifacts=root.parent / "artifacts", + keys=root.parent / "keys", + ), + backend=BackendConfig(), + cargo_cache=CargoCacheConfig(enabled=enabled, root=root, retention_days=retention_days), + ) + + +def _seed_entry(root: Path, digest: str, *, age_days: int, size: int = 1024) -> Path: + store = CargoCacheStore(root) + src = root.parent / f"seed-{digest}.tar.zst" + src.parent.mkdir(parents=True, exist_ok=True) + src.write_bytes(b"x" * size) + entry = store.publish(cache_key(lock_digest=digest, rustc_version="r", host_triple="t"), src) + stamp = time.time() - age_days * 86400 + import os + + os.utime(entry, (stamp, stamp)) + return entry + + +# ── cargo_cache_status ────────────────────────────────────────────────────── + + +def test_cargo_cache_status_missing_root(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + config = _make_config(root) + + status = cargo_cache_status(config) + + assert status == CacheStatus( + root=root, + total_bytes=0, + entry_count=0, + oldest_age_days=None, + retention_days=30, + ) + + +def test_cargo_cache_status_empty_root(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + root.mkdir(parents=True) + config = _make_config(root) + + status = cargo_cache_status(config) + + assert status.entry_count == 0 + assert status.total_bytes == 0 + assert status.oldest_age_days is None + + +def test_cargo_cache_status_populated_with_aged_entries(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + _seed_entry(root, "old", age_days=90, size=2048) + _seed_entry(root, "new", age_days=1, size=1024) + config = _make_config(root, retention_days=30) + + status = cargo_cache_status(config) + + assert status.entry_count == 2 + assert status.total_bytes == 2048 + 1024 + assert status.oldest_age_days is not None + assert status.oldest_age_days >= 89 + assert status.retention_days == 30 + assert status.root == root + + +def test_cargo_cache_status_ignores_incoming_staging_dir(tmp_path: Path) -> None: + """The ``.incoming`` staging directory used by ``publish()`` is not a real entry.""" + root = tmp_path / "cache" / "cargo" + _seed_entry(root, "one", age_days=1) + (root / ".incoming").mkdir(parents=True, exist_ok=True) + + config = _make_config(root) + status = cargo_cache_status(config) + + assert status.entry_count == 1 + + +def test_cargo_cache_status_reports_even_when_disabled(tmp_path: Path) -> None: + """Status is descriptive of disk state; ``enabled`` only gates purging.""" + root = tmp_path / "cache" / "cargo" + _seed_entry(root, "old", age_days=90) + config = _make_config(root, enabled=False) + + status = cargo_cache_status(config) + + assert status.entry_count == 1 + + +# ── purge_cargo_cache ──────────────────────────────────────────────────────── + + +def test_purge_cargo_cache_dry_run_counts_only_no_delete(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + old = _seed_entry(root, "old", age_days=90, size=3 * 1024 * 1024) + config = _make_config(root, retention_days=30) + + result = purge_cargo_cache(config, dry_run=True) + + assert result == PurgeResult(removed_entries=1, freed_bytes=3 * 1024 * 1024, dry_run=True) + assert old.exists(), "dry_run must not delete anything" + + +def test_purge_cargo_cache_real_purge_removes_only_old(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + old = _seed_entry(root, "old", age_days=90) + fresh = _seed_entry(root, "new", age_days=1) + config = _make_config(root, retention_days=30) + + result = purge_cargo_cache(config) + + assert result.removed_entries == 1 + assert result.dry_run is False + assert not old.exists() + assert fresh.exists() + + +def test_purge_cargo_cache_retention_override_wins_over_config(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + entry = _seed_entry(root, "mid", age_days=10) + config = _make_config(root, retention_days=30) # config alone would spare this entry + + result = purge_cargo_cache(config, retention_days=7) + + assert result.removed_entries == 1 + assert not entry.exists() + + +def test_purge_cargo_cache_disabled_returns_zero_and_does_not_touch_disk(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + old = _seed_entry(root, "old", age_days=90) + config = _make_config(root, enabled=False, retention_days=30) + + result = purge_cargo_cache(config) + + assert result == PurgeResult(removed_entries=0, freed_bytes=0, dry_run=False) + assert old.exists() + + +def test_purge_cargo_cache_nothing_to_purge(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + fresh = _seed_entry(root, "new", age_days=1) + config = _make_config(root, retention_days=30) + + result = purge_cargo_cache(config) + + assert result.removed_entries == 0 + assert result.freed_bytes == 0 + assert fresh.exists() + + +def test_purge_cargo_cache_missing_root_is_not_an_error(tmp_path: Path) -> None: + root = tmp_path / "cache" / "cargo" + config = _make_config(root, retention_days=30) + + result = purge_cargo_cache(config) + + assert result == PurgeResult(removed_entries=0, freed_bytes=0, dry_run=False) diff --git a/tests/python/test_operations_host.py b/tests/python/test_operations_host.py new file mode 100644 index 0000000..0b6df65 --- /dev/null +++ b/tests/python/test_operations_host.py @@ -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]