feat: B5 — port retention/template-backup to Python, deploy systemd timers

- Add `retention run` command (ports Invoke-RetentionPolicy.ps1): purges
  old artifact/log dirs with aggressive mode when disk space is low.
- Add `template backup` command (ports Backup-CITemplate.ps1): 7z -mx=1
  compressed archives in /var/lib/ci/backups/, prunes to keep-count=3,
  stops/restarts act-runner.service around the copy.
- Update ci-retention-policy.service and ci-backup-template.service to use
  Python; pwsh is no longer required on the Linux host.
- Fix ci-watch-runner-health.service: pass --service-name act-runner
  (Linux service name, not Windows act_runner default).
- Fix _list_orphans in vm.py: wrap is_dir() inside the OSError try block
  so a stat() failure on an entry is silently skipped rather than raised.
- Mark B5 complete in PhaseB-user-checklist.md.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
2026-05-21 20:35:07 +02:00
parent 536fd688e8
commit 43a69b82db
12 changed files with 974 additions and 147 deletions
+305
View File
@@ -0,0 +1,305 @@
"""Tests for ``ci_orchestrator template backup``."""
from __future__ import annotations
import os
import time
from pathlib import Path
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.template as tmpl_mod
from ci_orchestrator.__main__ import cli
# ── fixtures ──────────────────────────────────────────────────────────────────
@pytest.fixture()
def template_dir(tmp_path: Path) -> Path:
"""A fake template directory with a couple of files."""
t = tmp_path / "templates" / "WinBuild2025"
t.mkdir(parents=True)
(t / "WinBuild2025.vmx").write_text("vmx content")
(t / "WinBuild2025.vmdk").write_bytes(b"\x00" * 1024)
return t
@pytest.fixture()
def backup_dir(tmp_path: Path) -> Path:
d = tmp_path / "backups"
d.mkdir()
return d
def _fake_compress(source: Path, archive: Path) -> None:
"""Simulate 7z by writing a small placeholder file."""
archive.write_bytes(b"fake-7z-archive")
@pytest.fixture()
def _no_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
"""Prevent real systemctl and 7z calls in tests."""
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **_kw: False)
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **_kw: None)
monkeypatch.setattr(tmpl_mod, "_compress_7z", _fake_compress)
# ── basic backup ─────────────────────────────────────────────────────────────
def test_backup_creates_timestamped_archive(
template_dir: Path, backup_dir: Path, _no_systemctl: None
) -> None:
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--template-path",
str(template_dir),
"--backup-base-dir",
str(backup_dir),
"--skip-runner-stop",
],
)
assert result.exit_code == 0, result.output
archives = list(backup_dir.iterdir())
assert len(archives) == 1
assert archives[0].name.startswith("Template_")
assert archives[0].suffix == ".7z"
def test_backup_missing_template_raises(
backup_dir: Path, tmp_path: Path, _no_systemctl: None
) -> None:
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--template-path",
str(tmp_path / "nonexistent"),
"--backup-base-dir",
str(backup_dir),
"--skip-runner-stop",
],
)
assert result.exit_code != 0
assert "not found" in result.output
# ── --all-templates ───────────────────────────────────────────────────────────
def test_all_templates_backs_up_every_subdir(
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
) -> None:
tmpl_root = tmp_path / "templates"
for name in ("WinBuild2025", "WinBuild2022", "LinuxBuild2404"):
d = tmpl_root / name
d.mkdir(parents=True)
(d / f"{name}.vmx").write_text("vmx")
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--all-templates",
"--backup-base-dir",
str(backup_dir),
"--skip-runner-stop",
],
)
assert result.exit_code == 0, result.output
names = {f.name for f in backup_dir.iterdir() if f.suffix == ".7z"}
assert any("WinBuild2025" in n for n in names)
assert any("WinBuild2022" in n for n in names)
assert any("LinuxBuild2404" in n for n in names)
def test_all_templates_empty_dir_raises(
tmp_path: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch, _no_systemctl: None
) -> None:
tmpl_root = tmp_path / "templates"
tmpl_root.mkdir()
monkeypatch.setenv("CI_TEMPLATES", str(tmpl_root))
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--all-templates",
"--backup-base-dir",
str(backup_dir),
"--skip-runner-stop",
],
)
assert result.exit_code != 0
assert "No template" in result.output
# ── pruning ───────────────────────────────────────────────────────────────────
def test_prune_keeps_only_keep_count(
template_dir: Path, backup_dir: Path, _no_systemctl: None
) -> None:
for i in range(3):
old = backup_dir / f"Template_2025010{i}_120000.7z"
old.write_bytes(b"old-archive")
mtime = time.time() - (3 - i) * 3600
os.utime(old, (mtime, mtime))
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--template-path",
str(template_dir),
"--backup-base-dir",
str(backup_dir),
"--keep-count",
"2",
"--skip-runner-stop",
],
)
assert result.exit_code == 0, result.output
remaining = [f for f in backup_dir.iterdir() if f.suffix == ".7z"]
assert len(remaining) == 2, f"Expected 2 archives, got {len(remaining)}: {[f.name for f in remaining]}"
assert "Pruning" in result.output
# ── --what-if ────────────────────────────────────────────────────────────────
def test_what_if_no_copy(template_dir: Path, backup_dir: Path, _no_systemctl: None) -> None:
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--template-path",
str(template_dir),
"--backup-base-dir",
str(backup_dir),
"--skip-runner-stop",
"--what-if",
],
)
assert result.exit_code == 0, result.output
assert not list(backup_dir.iterdir()), "--what-if must not create anything"
assert "WhatIf" in result.output
# ── runner stop/start ─────────────────────────────────────────────────────────
def test_stop_runner_called_when_active(
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
stopped: list[bool] = []
started: list[bool] = []
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **kw: stopped.append(True) or True)
monkeypatch.setattr(tmpl_mod, "_start_runner", lambda **kw: started.append(True))
monkeypatch.setattr(tmpl_mod, "_compress_7z", _fake_compress)
result = CliRunner().invoke(
cli,
[
"template",
"backup",
"--template-path",
str(template_dir),
"--backup-base-dir",
str(backup_dir),
],
)
assert result.exit_code == 0, result.output
assert stopped, "_stop_runner should have been called"
assert started, "_start_runner should have been called"
def test_skip_runner_stop_skips_systemctl(
template_dir: Path, backup_dir: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
stopped: list[bool] = []
monkeypatch.setattr(tmpl_mod, "_stop_runner", lambda **kw: stopped.append(True) or True)
monkeypatch.setattr(tmpl_mod, "_compress_7z", _fake_compress)
CliRunner().invoke(
cli,
[
"template",
"backup",
"--template-path",
str(template_dir),
"--backup-base-dir",
str(backup_dir),
"--skip-runner-stop",
],
)
assert not stopped, "--skip-runner-stop should prevent _stop_runner from being called"
# ── unit helpers: _stop_runner / _start_runner ───────────────────────────────
def test_stop_runner_noop_when_inactive(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(
tmpl_mod.subprocess,
"run",
lambda cmd, **_kw: type("R", (), {"stdout": "inactive\n", "returncode": 0})(),
)
assert tmpl_mod._stop_runner(what_if=False) is False
def test_stop_runner_returns_true_when_active(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[list[str]] = []
def _fake_run(cmd: list[str], **_kw: object) -> object:
calls.append(cmd)
return type("R", (), {"stdout": "active\n", "returncode": 0})()
monkeypatch.setattr(tmpl_mod.subprocess, "run", _fake_run)
result = tmpl_mod._stop_runner(what_if=False)
assert result is True
assert any("stop" in c for c in calls)
def test_stop_runner_what_if_does_not_call_stop(monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[list[str]] = []
def _fake_run(cmd: list[str], **_kw: object) -> object:
calls.append(cmd)
return type("R", (), {"stdout": "active\n", "returncode": 0})()
monkeypatch.setattr(tmpl_mod.subprocess, "run", _fake_run)
result = tmpl_mod._stop_runner(what_if=True)
assert result is True
assert not any("stop" in c for c in calls), "what_if=True must not call systemctl stop"
# ── _compress_7z args ─────────────────────────────────────────────────────────
def test_compress_7z_uses_mx1(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
calls: list[list[str]] = []
def _fake_run(cmd: list[str], **_kw: object) -> object:
calls.append(cmd)
return type("R", (), {"returncode": 0})()
monkeypatch.setattr(tmpl_mod.subprocess, "run", _fake_run)
tmpl_mod._compress_7z(tmp_path / "src", tmp_path / "out.7z")
assert calls, "subprocess.run should have been called"
cmd = calls[0]
assert cmd[0] == "7z"
assert "-mx=1" in cmd