feat(web): dashboard redesign — Gitea palette, health panels, VM/audit enrichment (stages A–C)

Stage A: swap CSS tokens to the real Gitea dark palette (accent blue #4183c4); add Jinja formatting filters (humanbytes/duration/relative_time/localtime/shorten_path/thousands), a health status-header, and an inline SVG icon sprite; fix two UI bugs — unthemed action pages (htmx swap into #action-slot with a full-page HX-Request fallback) and polling wiping form input (hx-preserve).

Stage B: four read-only panels that degrade to notices (never 500) — host resources (/proc, two-sample CPU%), VMware module health, template inventory (.vmem warn + last backup), and a path-contained file browser; backup card becomes a validated template dropdown; expensive dir-size walk is TTL-cached. Adds operations/inventory.py plus host_resources()/vmware_module_health() in operations/host.py.

Stage C: parse_clone_name() adds a Job column (job_id + started) to the VM table; nav gains a Gitea link-out derived from the OIDC issuer origin (no fabricated run URLs); audit page gets action icons, relative time, and a client-side filter.

Gates: 989 tests pass, coverage 94.12%, ruff + mypy --strict clean. Stage D (service management) intentionally deferred.
This commit is contained in:
2026-07-24 11:46:50 +02:00
parent 907e79818e
commit 9d7bbee35d
33 changed files with 4099 additions and 65 deletions
+70 -1
View File
@@ -9,7 +9,7 @@ from __future__ import annotations
from pathlib import Path
from ci_orchestrator.config import WebConfig, WebOidcConfig, load_config
from ci_orchestrator.config import WebConfig, WebOidcConfig, gitea_base_url, load_config
def _paths_section() -> str:
@@ -182,3 +182,72 @@ def test_toml_secret_looking_value_is_never_surfaced_as_a_secret(tmp_path: Path)
# No attribute on WebOidcConfig can ever hold "should-never-be-read".
for value in vars(cfg.web.oidc).values():
assert value != "should-never-be-read"
# ── [web].gitea_url (Stage C2) ───────────────────────────────────────────────
def test_gitea_url_absent_from_toml_defaults_to_none() -> None:
assert WebConfig().gitea_url is None
def test_gitea_url_explicit_toml_value_populates_field(tmp_path: Path) -> None:
extra = '\n[web]\ngitea_url = "https://gitea.example.lan"\n'
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
assert cfg.web.gitea_url == "https://gitea.example.lan"
def test_gitea_base_url_explicit_override_wins_over_issuer(tmp_path: Path) -> None:
extra = (
"\n[web]\n"
'gitea_url = "https://gitea.explicit.lan"\n'
"\n[web.oidc]\n"
'issuer = "https://gitea.emulab.it"\n'
'client_id = "ci-web"\n'
'redirect_url = "http://192.168.1.50:8800/auth/callback"\n'
)
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
assert gitea_base_url(cfg.web) == "https://gitea.explicit.lan"
def test_gitea_base_url_derives_origin_from_oidc_issuer(tmp_path: Path) -> None:
extra = (
"\n[web.oidc]\n"
'issuer = "https://gitea.emulab.it"\n'
'client_id = "ci-web"\n'
'redirect_url = "http://192.168.1.50:8800/auth/callback"\n'
)
cfg = load_config(config_path=_write_toml(tmp_path, extra), env={})
assert cfg.web.gitea_url is None # not set explicitly
assert gitea_base_url(cfg.web) == "https://gitea.emulab.it"
def test_gitea_base_url_strips_issuer_path_and_query() -> None:
web = WebConfig(
oidc=WebOidcConfig(
issuer="https://gitea.emulab.it/some/path?query=1",
client_id="ci-web",
redirect_url="http://127.0.0.1:8800/auth/callback",
)
)
assert gitea_base_url(web) == "https://gitea.emulab.it"
def test_gitea_base_url_none_when_oidc_unset_and_no_explicit_url() -> None:
assert gitea_base_url(WebConfig()) is None
def test_gitea_base_url_none_when_issuer_unparseable() -> None:
web = WebConfig(
oidc=WebOidcConfig(
issuer="not-a-url",
client_id="ci-web",
redirect_url="http://127.0.0.1:8800/auth/callback",
)
)
assert gitea_base_url(web) is None
+244
View File
@@ -0,0 +1,244 @@
"""Tests for the Stage B (B1/B2) ``operations.host`` additions:
``host_resources`` and ``vmware_module_health``.
Everything reads through the ``_read_proc`` seam -- monkeypatched per test to
fixture strings, never a real ``/proc``. ``time.sleep`` is also patched to a
no-op so the two-sample CPU read never actually waits ``cpu_sample_ms``.
"""
from __future__ import annotations
from typing import Any
import pytest
import ci_orchestrator.operations.host as host_module
from ci_orchestrator.operations.host import (
HostResources,
RunnerStatus,
VmwareModuleHealth,
host_resources,
vmware_module_health,
)
# ── fixture /proc/stat samples ──────────────────────────────────────────────
# Columns: user nice system idle iowait irq softirq steal guest guest_nice.
# Sample 1: idle=8000+0=8000, total=1000+0+1000+8000=10000
# Sample 2: idle=8200+0=8200, total=1100+0+1100+8200=10400
# idle_delta=200, total_delta=400 -> cpu_pct = 100*(1 - 200/400) = 50.0
_STAT_1 = "cpu 1000 0 1000 8000 0 0 0 0 0 0\ncpu0 1000 0 1000 8000 0 0 0 0 0 0\n"
_STAT_2 = "cpu 1100 0 1100 8200 0 0 0 0 0 0\ncpu0 1100 0 1100 8200 0 0 0 0 0 0\n"
_MEMINFO = (
"MemTotal: 16384000 kB\n"
"MemFree: 2000000 kB\n"
"MemAvailable: 4096000 kB\n"
"Buffers: 100000 kB\n"
)
# total_gb = 16384000/1024**2 = 15.625; used = (16384000-4096000)/1024**2 = 11.71875
_UPTIME = "123456.78 987654.32\n"
_LOADAVG = "0.50 0.75 1.00 3/456 7890\n"
def _fake_reader(calls: dict[str, int]) -> Any:
def _read(path: str) -> str:
calls[path] = calls.get(path, 0) + 1
if path == "/proc/stat":
return _STAT_1 if calls[path] == 1 else _STAT_2
if path == "/proc/meminfo":
return _MEMINFO
if path == "/proc/uptime":
return _UPTIME
if path == "/proc/loadavg":
return _LOADAVG
raise AssertionError(f"unexpected proc path: {path}")
return _read
def _install_fake_proc(monkeypatch: pytest.MonkeyPatch) -> dict[str, int]:
calls: dict[str, int] = {}
monkeypatch.setattr(host_module, "_read_proc", _fake_reader(calls))
monkeypatch.setattr(host_module.time, "sleep", lambda _seconds: None)
return calls
# ── host_resources: happy path, two-sample CPU math ─────────────────────────
def test_host_resources_two_sample_cpu_math(monkeypatch: pytest.MonkeyPatch) -> None:
calls = _install_fake_proc(monkeypatch)
monkeypatch.setattr(host_module.os, "cpu_count", lambda: 8)
result = host_resources(cpu_sample_ms=150)
assert result.cpu_pct == pytest.approx(50.0)
assert result.ncpu == 8
assert calls["/proc/stat"] == 2 # exactly two samples, per the B1 spec
def test_host_resources_reads_proc_stat_twice_with_a_sleep_between(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls = _install_fake_proc(monkeypatch)
sleeps: list[float] = []
monkeypatch.setattr(host_module.time, "sleep", lambda seconds: sleeps.append(seconds))
host_resources(cpu_sample_ms=200)
assert calls["/proc/stat"] == 2
assert sleeps == [pytest.approx(0.2)]
def test_host_resources_mem_used_is_total_minus_available(
monkeypatch: pytest.MonkeyPatch,
) -> None:
_install_fake_proc(monkeypatch)
result = host_resources()
assert result.mem_total_gb == pytest.approx(16384000 / 1024**2)
assert result.mem_used_gb == pytest.approx((16384000 - 4096000) / 1024**2)
def test_host_resources_uptime_and_loadavg(monkeypatch: pytest.MonkeyPatch) -> None:
_install_fake_proc(monkeypatch)
result = host_resources()
assert result.uptime_sec == 123456
assert result.load1 == pytest.approx(0.50)
assert result.load5 == pytest.approx(0.75)
assert result.load15 == pytest.approx(1.00)
# ── host_resources: degrade to zeros on OSError, never raise ────────────────
def test_host_resources_degrades_to_zeros_on_oserror(monkeypatch: pytest.MonkeyPatch) -> None:
def _boom(_path: str) -> str:
raise OSError("no /proc on this platform")
monkeypatch.setattr(host_module, "_read_proc", _boom)
monkeypatch.setattr(host_module.os, "cpu_count", lambda: 4)
result = host_resources()
assert result == HostResources(
cpu_pct=0.0,
ncpu=4,
mem_total_gb=0.0,
mem_used_gb=0.0,
uptime_sec=0,
load1=0.0,
load5=0.0,
load15=0.0,
)
def test_host_resources_degrades_to_zeros_when_second_stat_sample_fails(
monkeypatch: pytest.MonkeyPatch,
) -> None:
"""The first ``/proc/stat`` read can succeed and the second still fail."""
calls = {"n": 0}
def _flaky(path: str) -> str:
if path != "/proc/stat":
raise AssertionError("should not reach meminfo after stat blows up")
calls["n"] += 1
if calls["n"] == 1:
return _STAT_1
raise OSError("gone")
monkeypatch.setattr(host_module, "_read_proc", _flaky)
monkeypatch.setattr(host_module.time, "sleep", lambda _s: None)
result = host_resources()
assert result.cpu_pct == 0.0
assert result.uptime_sec == 0
# ── vmware_module_health: /proc/modules parsing ─────────────────────────────
def _fake_runner_status(statuses: list[RunnerStatus]) -> Any:
def _query(_names: list[str]) -> list[RunnerStatus]:
return statuses
return _query
def test_vmware_module_health_both_modules_loaded(monkeypatch: pytest.MonkeyPatch) -> None:
modules_text = "vmmon 57344 1 - Live 0x0000000000000000\nvmnet 69632 3 - Live 0x0000000000000000\n"
monkeypatch.setattr(host_module, "_read_proc", lambda _path: modules_text)
monkeypatch.setattr(
host_module,
"runner_status",
_fake_runner_status([RunnerStatus(name="vmware-host.service", active=True, sub_state="running")]),
)
result = vmware_module_health()
assert result == VmwareModuleHealth(
vmmon_loaded=True,
vmnet_loaded=True,
host_service_active=True,
host_service_sub_state="running",
)
def test_vmware_module_health_modules_absent(monkeypatch: pytest.MonkeyPatch) -> None:
modules_text = "usbcore 300000 4 - Live 0x0000000000000000\n"
monkeypatch.setattr(host_module, "_read_proc", lambda _path: modules_text)
monkeypatch.setattr(
host_module,
"runner_status",
_fake_runner_status([RunnerStatus(name="vmware-host.service", active=False, sub_state="dead")]),
)
result = vmware_module_health()
assert result.vmmon_loaded is False
assert result.vmnet_loaded is False
assert result.host_service_active is False
assert result.host_service_sub_state == "dead"
def test_vmware_module_health_degrades_on_proc_modules_oserror(
monkeypatch: pytest.MonkeyPatch,
) -> None:
def _boom(_path: str) -> str:
raise OSError("no /proc/modules")
monkeypatch.setattr(host_module, "_read_proc", _boom)
monkeypatch.setattr(
host_module,
"runner_status",
_fake_runner_status([RunnerStatus(name="vmware-host.service", active=True, sub_state="running")]),
)
result = vmware_module_health()
assert result.vmmon_loaded is False
assert result.vmnet_loaded is False
# A /proc/modules failure must not also blank the (independent) service query.
assert result.host_service_active is True
def test_vmware_module_health_degrades_when_runner_status_raises(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(host_module, "_read_proc", lambda _path: "vmmon 1 1 - Live 0x0\n")
def _boom(_names: list[str]) -> list[RunnerStatus]:
raise RuntimeError("systemctl vanished")
monkeypatch.setattr(host_module, "runner_status", _boom)
result = vmware_module_health()
assert result.vmmon_loaded is True
assert result.host_service_active is False
assert result.host_service_sub_state == "unknown"
+503
View File
@@ -0,0 +1,503 @@
"""Tests for the Stage B (B3/B4) ``operations.inventory`` module.
Template discovery (``list_templates``) and containment-checked directory
listings (``list_dir``/``list_backups``/``list_artifacts``). No real ``7z``/
``vmrun`` involved -- everything is plain files under ``tmp_path``.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
import pytest
import ci_orchestrator.operations.inventory as inventory_module
from ci_orchestrator.config import BackendConfig, Config, Paths
from ci_orchestrator.operations.inventory import (
FileEntry,
TemplateInfo,
list_artifacts,
list_backups,
list_dir,
list_templates,
)
def _make_config(root: Path) -> Config:
return Config(
paths=Paths(
root=root,
templates=root / "templates",
build_vms=root / "build-vms",
artifacts=root / "artifacts",
keys=root / "keys",
),
backend=BackendConfig(),
)
@pytest.fixture(autouse=True)
def _clear_dir_bytes_cache() -> None:
"""The B7 ``dir_bytes`` cache is module-level -- start each test clean."""
inventory_module._dir_bytes_cache.clear()
# ── list_templates ───────────────────────────────────────────────────────────
def _write_vmx(path: Path, guest_os: str) -> None:
path.write_text(f'.encoding = "UTF-8"\nguestOS = "{guest_os}"\n', encoding="utf-8")
def test_list_templates_missing_dir_is_empty(tmp_path: Path) -> None:
config = _make_config(tmp_path) # templates dir never created
assert list_templates(config) == []
def test_list_templates_parses_guest_os_and_transport_ssh(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "LinuxBuild2404"
tmpl_dir.mkdir(parents=True)
_write_vmx(tmpl_dir / "LinuxBuild2404.vmx", "ubuntu-64")
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.name == "LinuxBuild2404"
assert info.vmx == tmpl_dir / "LinuxBuild2404.vmx"
assert info.guest_os == "ubuntu-64"
assert info.transport == "ssh"
def test_list_templates_windows_guest_os_is_winrm_transport(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "WinBuild2025"
tmpl_dir.mkdir(parents=True)
_write_vmx(tmpl_dir / "WinBuild2025.vmx", "windows9srv-64")
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.transport == "winrm"
def test_list_templates_no_vmx_is_unknown_transport(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "Empty"
tmpl_dir.mkdir(parents=True)
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.vmx is None
assert info.guest_os is None
assert info.transport == "unknown"
def test_list_templates_has_vmem_true_when_present(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "WithVmem"
tmpl_dir.mkdir(parents=True)
_write_vmx(tmpl_dir / "WithVmem.vmx", "ubuntu-64")
(tmpl_dir / "WithVmem.vmem").write_bytes(b"\x00" * 16)
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.has_vmem is True
def test_list_templates_has_vmem_false_when_absent(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "Clean"
tmpl_dir.mkdir(parents=True)
_write_vmx(tmpl_dir / "Clean.vmx", "ubuntu-64")
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.has_vmem is False
def test_list_templates_dir_bytes_sums_file_sizes(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "Sized"
nested = tmpl_dir / "nested"
nested.mkdir(parents=True)
(tmpl_dir / "a.vmdk").write_bytes(b"x" * 100)
(nested / "b.vmdk").write_bytes(b"y" * 50)
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.dir_bytes == 150
def test_list_templates_multiple_entries_sorted_by_name(tmp_path: Path) -> None:
for name in ("WinBuild2022", "LinuxBuild2404", "WinBuild2025"):
(tmp_path / "templates" / name).mkdir(parents=True)
config = _make_config(tmp_path)
names = [t.name for t in list_templates(config)]
assert names == sorted(names)
def test_list_templates_last_backup_picks_newest_matching_archive(tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "LinuxBuild2404"
tmpl_dir.mkdir(parents=True)
backups_dir = tmp_path / "backups"
backups_dir.mkdir()
older = backups_dir / "Template_LinuxBuild2404_20260101_000000.7z"
newer = backups_dir / "Template_LinuxBuild2404_20260601_000000.7z"
unrelated = backups_dir / "Template_OtherTemplate_20260701_000000.7z"
older.write_bytes(b"old")
unrelated.write_bytes(b"other")
time.sleep(0.01)
newer.write_bytes(b"new")
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.last_backup is not None
assert info.last_backup.name == "Template_LinuxBuild2404_20260601_000000.7z"
def test_list_templates_last_backup_none_when_no_backups_dir(tmp_path: Path) -> None:
(tmp_path / "templates" / "LinuxBuild2404").mkdir(parents=True)
config = _make_config(tmp_path)
[info] = list_templates(config)
assert info.last_backup is None
def test_list_templates_skips_regular_files_directly_under_templates(tmp_path: Path) -> None:
templates_dir = tmp_path / "templates"
templates_dir.mkdir(parents=True)
(templates_dir / "README.txt").write_text("not a template", encoding="utf-8")
(templates_dir / "Real").mkdir()
config = _make_config(tmp_path)
names = [t.name for t in list_templates(config)]
assert names == ["Real"]
def test_list_templates_dir_bytes_is_cached(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
tmpl_dir = tmp_path / "templates" / "Cached"
tmpl_dir.mkdir(parents=True)
(tmpl_dir / "a.vmdk").write_bytes(b"x" * 10)
config = _make_config(tmp_path)
walk_calls: list[Path] = []
real_walk = os.walk
def _counting_walk(path: object, *a: object, **kw: object) -> object:
walk_calls.append(Path(str(path)))
return real_walk(path, *a, **kw) # type: ignore[arg-type]
monkeypatch.setattr(inventory_module.os, "walk", _counting_walk)
first = list_templates(config)[0].dir_bytes
second = list_templates(config)[0].dir_bytes
assert first == second == 10
assert len(walk_calls) == 1, "second call must hit the TTL cache, not os.walk again"
# ── list_dir / containment ───────────────────────────────────────────────────
def test_list_dir_accepts_root_itself(tmp_path: Path) -> None:
(tmp_path / "f1.txt").write_bytes(b"a")
entries = list_dir(tmp_path, root=tmp_path)
assert [e.name for e in entries] == ["f1.txt"]
def test_list_dir_accepts_real_subdir(tmp_path: Path) -> None:
sub = tmp_path / "backups"
sub.mkdir()
(sub / "f1.7z").write_bytes(b"a")
entries = list_dir(sub, root=tmp_path, pattern="*.7z")
assert [e.name for e in entries] == ["f1.7z"]
def test_list_dir_rejects_base_outside_root(tmp_path: Path) -> None:
root = tmp_path / "root"
root.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
(outside / "secret.txt").write_bytes(b"nope")
assert list_dir(outside, root=root) == []
def test_list_dir_rejects_dotdot_escape(tmp_path: Path) -> None:
root = tmp_path / "root"
root.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
(outside / "secret.txt").write_bytes(b"nope")
escaping = root / ".." / "outside"
assert list_dir(escaping, root=root) == []
def test_list_dir_rejects_symlink_escaping_root(tmp_path: Path) -> None:
root = tmp_path / "root"
root.mkdir()
outside = tmp_path / "outside"
outside.mkdir()
(outside / "secret.txt").write_bytes(b"nope")
link = root / "escape_link"
link.symlink_to(outside, target_is_directory=True)
assert list_dir(link, root=root) == []
def test_list_dir_missing_base_is_empty(tmp_path: Path) -> None:
assert list_dir(tmp_path / "does-not-exist", root=tmp_path) == []
def test_list_dir_sorted_newest_first(tmp_path: Path) -> None:
older = tmp_path / "a.txt"
newer = tmp_path / "b.txt"
older.write_bytes(b"old")
time.sleep(0.01)
newer.write_bytes(b"new")
entries = list_dir(tmp_path, root=tmp_path)
assert [e.name for e in entries] == ["b.txt", "a.txt"]
def test_list_dir_pattern_filters_extension(tmp_path: Path) -> None:
(tmp_path / "keep.7z").write_bytes(b"x")
(tmp_path / "skip.txt").write_bytes(b"y")
entries = list_dir(tmp_path, root=tmp_path, pattern="*.7z")
assert [e.name for e in entries] == ["keep.7z"]
def test_list_dir_entry_shape(tmp_path: Path) -> None:
f = tmp_path / "x.7z"
f.write_bytes(b"12345")
[entry] = list_dir(tmp_path, root=tmp_path)
assert isinstance(entry, FileEntry)
assert entry.size_bytes == 5
assert entry.mtime == pytest.approx(f.stat().st_mtime)
# ── list_backups / list_artifacts ────────────────────────────────────────────
def test_list_backups_reads_root_backups_dir(tmp_path: Path) -> None:
backups = tmp_path / "backups"
backups.mkdir()
(backups / "Template_X_20260101_000000.7z").write_bytes(b"x")
(backups / "notes.txt").write_bytes(b"ignored")
config = _make_config(tmp_path)
entries = list_backups(config)
assert [e.name for e in entries] == ["Template_X_20260101_000000.7z"]
def test_list_backups_missing_dir_is_empty(tmp_path: Path) -> None:
config = _make_config(tmp_path)
assert list_backups(config) == []
def test_list_artifacts_reads_configured_artifacts_dir(tmp_path: Path) -> None:
config = _make_config(tmp_path)
config.paths.artifacts.mkdir(parents=True)
(config.paths.artifacts / "build.log").write_bytes(b"log")
entries = list_artifacts(config)
assert [e.name for e in entries] == ["build.log"]
def test_list_artifacts_outside_root_is_rejected(tmp_path: Path) -> None:
"""``config.paths.artifacts`` pointed outside ``root`` must still 0-list.
``outside_artifacts`` sits beside ``root`` (both under ``tmp_path``) --
a sibling directory, not a descendant -- so containment must reject it.
"""
root = tmp_path / "root"
root.mkdir()
outside_artifacts = tmp_path / "artifacts-outside"
outside_artifacts.mkdir()
(outside_artifacts / "build.log").write_bytes(b"log")
config = Config(
paths=Paths(
root=root,
templates=root / "templates",
build_vms=root / "build-vms",
artifacts=outside_artifacts,
keys=root / "keys",
),
backend=BackendConfig(),
)
assert list_artifacts(config) == []
# ── defensive OSError degrade paths ("never raise") ─────────────────────────
class _BoomIterdir:
"""Duck-typed stand-in for a ``Path`` whose ``iterdir()`` blows up."""
def iterdir(self) -> list[Path]:
raise OSError("denied")
class _BoomGlob:
"""Duck-typed stand-in for a ``Path`` whose ``glob()`` blows up."""
def glob(self, _pattern: str) -> list[Path]:
raise OSError("denied")
class _BoomReadText:
"""Duck-typed stand-in for a ``Path`` whose ``read_text()`` blows up."""
def read_text(self, encoding: str = "utf-8", errors: str = "strict") -> str:
raise OSError("denied")
class _BoomResolve:
"""Duck-typed stand-in for a ``Path`` whose ``resolve()`` blows up."""
def resolve(self) -> Path:
raise OSError("denied")
def test_find_vmx_oserror_returns_none() -> None:
assert inventory_module._find_vmx(_BoomIterdir()) is None # type: ignore[arg-type]
def test_has_vmem_oserror_returns_false() -> None:
assert inventory_module._has_vmem(_BoomGlob()) is False # type: ignore[arg-type]
def test_read_guest_os_oserror_returns_empty_string() -> None:
assert inventory_module._read_guest_os(_BoomReadText()) == "" # type: ignore[arg-type]
def test_dir_bytes_cached_resolve_oserror_returns_zero() -> None:
assert inventory_module._dir_bytes_cached(_BoomResolve()) == 0 # type: ignore[arg-type]
def test_newest_backup_glob_oserror_returns_none(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
backups_dir = tmp_path / "backups"
backups_dir.mkdir()
monkeypatch.setattr(Path, "glob", lambda self, pattern: (_ for _ in ()).throw(OSError("denied")))
assert inventory_module._newest_backup(backups_dir, "X") is None
def test_dir_bytes_cached_walk_oserror_degrades_to_zero(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
tmpl_dir = tmp_path / "T"
tmpl_dir.mkdir()
(tmpl_dir / "a.txt").write_bytes(b"x" * 5)
def _boom_walk(_path: Path, *_a: object, **_kw: object) -> object:
raise OSError("walk exploded")
monkeypatch.setattr(inventory_module.os, "walk", _boom_walk)
assert inventory_module._dir_bytes_cached(tmpl_dir) == 0
def test_dir_bytes_cache_recomputes_after_ttl_expiry(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
"""B7: a cache entry older than the TTL must be recomputed, not reused forever."""
tmpl_dir = tmp_path / "templates" / "Stale"
tmpl_dir.mkdir(parents=True)
(tmpl_dir / "a.vmdk").write_bytes(b"x" * 10)
config = _make_config(tmp_path)
clock = {"t": 0.0}
monkeypatch.setattr(inventory_module.time, "monotonic", lambda: clock["t"])
walk_calls: list[Path] = []
real_walk = os.walk
def _counting_walk(path: object, *a: object, **kw: object) -> object:
walk_calls.append(Path(str(path)))
return real_walk(path, *a, **kw) # type: ignore[arg-type]
monkeypatch.setattr(inventory_module.os, "walk", _counting_walk)
first = list_templates(config)[0].dir_bytes
clock["t"] += inventory_module._DIR_BYTES_TTL_SECONDS + 1.0
second = list_templates(config)[0].dir_bytes
assert first == second == 10
assert len(walk_calls) == 2, "an expired cache entry must trigger a fresh os.walk"
def test_list_templates_iterdir_oserror_returns_empty(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
templates_dir = tmp_path / "templates"
templates_dir.mkdir()
config = _make_config(tmp_path)
original_iterdir = Path.iterdir
def _boom_iterdir(self: Path) -> object:
if self == templates_dir:
raise OSError("denied")
return original_iterdir(self)
monkeypatch.setattr(Path, "iterdir", _boom_iterdir)
assert list_templates(config) == []
def test_list_dir_resolve_oserror_returns_empty(
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
) -> None:
monkeypatch.setattr(Path, "resolve", lambda self: (_ for _ in ()).throw(OSError("denied")))
assert list_dir(tmp_path, root=tmp_path) == []
def test_list_dir_skips_subdirectories(tmp_path: Path) -> None:
(tmp_path / "subdir").mkdir()
(tmp_path / "file.txt").write_bytes(b"x")
entries = list_dir(tmp_path, root=tmp_path)
assert [e.name for e in entries] == ["file.txt"]
def test_template_info_is_frozen_dataclass() -> None:
info = TemplateInfo(
name="X",
vmx=None,
guest_os=None,
transport="unknown",
dir_bytes=0,
has_vmem=False,
last_backup=None,
)
with pytest.raises(AttributeError):
info.name = "Y" # type: ignore[misc]
+59 -1
View File
@@ -9,7 +9,7 @@ coverage for ``list_live_vms`` and ``remove_vm``.
from __future__ import annotations
import time
from datetime import datetime, timedelta
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Any
@@ -18,13 +18,16 @@ import pytest
import ci_orchestrator.operations.vms as vms_module
from ci_orchestrator.backends.errors import BackendOperationFailed
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.commands.job import _make_clone_name
from ci_orchestrator.config import BackendConfig, Config, Paths
from ci_orchestrator.operations.vms import (
Orphan,
ParsedClone,
RemovalResult,
cleanup_orphans,
list_live_vms,
list_orphans,
parse_clone_name,
remove_vm,
)
@@ -506,3 +509,58 @@ def test_reconcile_ip_leases_unreadable_base_is_noop(tmp_path: Path) -> None:
leases = json.loads((tmp_path / "ip-pool.json").read_text())
assert leases["10.0.0.1"] == "clone-dead"
# ── parse_clone_name (Stage C1) ───────────────────────────────────────────
def test_parse_clone_name_round_trips_real_make_clone_name() -> None:
"""A real ``_make_clone_name`` output must invert cleanly."""
before = datetime.now(tz=UTC).replace(microsecond=0)
name = _make_clone_name("build-42")
after = datetime.now(tz=UTC).replace(microsecond=0)
parsed = parse_clone_name(name)
assert parsed.job_id == "build-42"
assert parsed.started is not None
# started is a naive datetime built from UTC wall-clock digits -- compare
# against naive bounds taken from an aware UTC "now" either side of the call.
assert before.replace(tzinfo=None) <= parsed.started <= after.replace(tzinfo=None) + timedelta(seconds=1)
def test_parse_clone_name_job_id_containing_underscore() -> None:
parsed = parse_clone_name("Clone_my_job_42_20250101_123456_abc123")
assert parsed.job_id == "my_job_42"
assert parsed.started == datetime(2025, 1, 1, 12, 34, 56)
def test_parse_clone_name_non_clone_prefix_degrades_both_to_none() -> None:
assert parse_clone_name("SomeOtherVMName") == ParsedClone(job_id=None, started=None)
def test_parse_clone_name_too_few_parts_degrades_both_to_none() -> None:
# No room for date + time + suffix at all.
assert parse_clone_name("Clone_short") == ParsedClone(job_id=None, started=None)
def test_parse_clone_name_garbage_timestamp_degrades_started_only() -> None:
parsed = parse_clone_name("Clone_myjob_notadate_notatime_abc123")
assert parsed.job_id == "myjob"
assert parsed.started is None
def test_parse_clone_name_empty_job_id_degrades_to_none() -> None:
# No text between "Clone_" and the timestamp.
parsed = parse_clone_name("Clone_20250101_123456_abc123")
assert parsed.job_id is None
assert parsed.started == datetime(2025, 1, 1, 12, 34, 56)
def test_parse_clone_name_wrong_length_date_degrades_started_only() -> None:
parsed = parse_clone_name("Clone_job1_2025010_123456_abc123")
assert parsed.started is None
+10 -5
View File
@@ -759,7 +759,11 @@ def test_host_backup_missing_template_field_is_400(tmp_path: Path) -> None:
assert resp.status_code == 400
def test_host_backup_preview_missing_template_dir_is_error_partial(tmp_path: Path) -> None:
def test_host_backup_unknown_template_name_is_400_not_audited(tmp_path: Path) -> None:
"""B6: a template name absent from ``list_templates()`` is rejected before
``backup_template`` is ever called -- a pure validation refusal, so (like
the VM-remove path-containment check) it is never audited.
"""
config = _make_config(tmp_path)
app = create_app(config)
client = TestClient(app)
@@ -767,10 +771,8 @@ def test_host_backup_preview_missing_template_dir_is_error_partial(tmp_path: Pat
resp = client.post("/host/backup", data={"csrf": csrf, "template": "NoSuchTemplate"})
assert resp.status_code == 200
assert "error" in resp.text.lower()
entry = _last_audit_entry(config.web.audit_log)
assert entry.result == "error"
assert resp.status_code == 400
assert not config.web.audit_log.exists()
def test_host_backup_preview_then_execute_full_cycle(
@@ -812,6 +814,9 @@ def test_host_backup_execute_exception_never_500s(
raise RuntimeError("disk exploded")
monkeypatch.setattr(routes_module, "backup_template", _boom)
# Must be a name list_templates() actually discovers, or the B6
# known-template check 400s before backup_template is ever reached.
(tmp_path / "templates" / "X").mkdir(parents=True)
config = _make_config(tmp_path)
app = create_app(config)
client = TestClient(app)
+139
View File
@@ -20,6 +20,9 @@ import ci_orchestrator.web.app as app_module
from ci_orchestrator.backends.errors import BackendNotAvailable
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.config import BackendConfig, CargoCacheConfig, Config, Paths
from ci_orchestrator.operations.cache import CacheStatus
from ci_orchestrator.operations.host import DiskStatus, RunnerStatus
from ci_orchestrator.operations.vms import LiveVm
from ci_orchestrator.web.app import create_app, get_backend
_DiskUsage = namedtuple("_DiskUsage", ["total", "used", "free"])
@@ -268,3 +271,139 @@ def test_static_mount_serves_app_css(tmp_path: Path) -> None:
resp = client.get("/static/app.css")
assert resp.status_code == 200
# ── _health_summary (A3) ─────────────────────────────────────────────────
_ACTIVE_RUNNER = RunnerStatus(name="act-runner.service", active=True, sub_state="running")
_DEAD_RUNNER = RunnerStatus(name="act-runner-rust.service", active=False, sub_state="dead")
_LIVE_VM = LiveVm(
name="job1", vmx="/x/job1.vmx", ip="10.0.0.5", guest_os="ubuntu-64",
uptime_sec=120, is_orphan=False,
)
_ORPHAN_VM = LiveVm(
name="job2", vmx="/x/job2.vmx", ip=None, guest_os="ubuntu-64",
uptime_sec=99999, is_orphan=True,
)
_CACHE = CacheStatus(
root=Path("/cache"), total_bytes=1024, entry_count=1, oldest_age_days=1, retention_days=30
)
def test_health_summary_backend_unavailable_is_vm_unknown() -> None:
health = app_module._health_summary(
[], backend_available=False, cache=None, disk=None, runners=[]
)
assert health.vm_class == "tile-unknown"
assert health.vm_label == "unknown"
def test_health_summary_orphan_present_is_vm_warn() -> None:
health = app_module._health_summary(
[_LIVE_VM, _ORPHAN_VM], backend_available=True, cache=None, disk=None, runners=[]
)
assert health.vm_count == 2
assert health.orphan_count == 1
assert health.vm_class == "tile-warn"
assert health.vm_label == "orphaned"
def test_health_summary_no_orphans_is_vm_ok() -> None:
health = app_module._health_summary(
[_LIVE_VM], backend_available=True, cache=None, disk=None, runners=[]
)
assert health.vm_class == "tile-ok"
assert health.vm_label == "ok"
def test_health_summary_no_runners_is_unknown() -> None:
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=None, runners=[]
)
assert health.runners_total == 0
assert health.runner_class == "tile-unknown"
def test_health_summary_all_runners_active_is_ok() -> None:
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=None, runners=[_ACTIVE_RUNNER]
)
assert health.runners_active == 1
assert health.runners_total == 1
assert health.runner_class == "tile-ok"
def test_health_summary_a_down_runner_is_bad() -> None:
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=None,
runners=[_ACTIVE_RUNNER, _DEAD_RUNNER],
)
assert health.runners_active == 1
assert health.runners_total == 2
assert health.runner_class == "tile-bad"
assert health.runner_label == "down"
def test_health_summary_disk_none_is_unknown() -> None:
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=None, runners=[]
)
assert health.disk_used_pct is None
assert health.disk_class == "tile-unknown"
def test_health_summary_disk_above_critical_is_bad() -> None:
disk = DiskStatus(path=Path("/"), total_gb=100.0, free_gb=5.0, used_pct=90.0)
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=disk, runners=[]
)
assert health.disk_class == "tile-bad"
assert health.disk_label == "critical"
def test_health_summary_disk_above_elevated_is_warn() -> None:
disk = DiskStatus(path=Path("/"), total_gb=100.0, free_gb=25.0, used_pct=75.0)
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=disk, runners=[]
)
assert health.disk_class == "tile-warn"
assert health.disk_label == "elevated"
def test_health_summary_disk_healthy_is_ok() -> None:
disk = DiskStatus(path=Path("/"), total_gb=100.0, free_gb=60.0, used_pct=40.0)
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=disk, runners=[]
)
assert health.disk_class == "tile-ok"
assert health.disk_label == "ok"
def test_health_summary_cache_none_is_unknown() -> None:
health = app_module._health_summary(
[], backend_available=True, cache=None, disk=None, runners=[]
)
assert health.cache_bytes is None
assert health.cache_class == "tile-unknown"
def test_health_summary_cache_present_is_ok() -> None:
health = app_module._health_summary(
[], backend_available=True, cache=_CACHE, disk=None, runners=[]
)
assert health.cache_bytes == 1024
assert health.cache_class == "tile-ok"
def test_dashboard_includes_status_header_tiles(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_quiet_host(monkeypatch)
app = create_app(_make_config(tmp_path))
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
client = TestClient(app)
resp = client.get("/")
assert resp.status_code == 200
assert 'id="status-header"' in resp.text
+242
View File
@@ -0,0 +1,242 @@
"""Tests for the Stage A (A2) Jinja display filters.
Pure-function tests -- no app/template rendering needed, just exact-string
assertions on each filter's chosen output format (the brief leaves the
precise text up to the implementation, as long as it's exercised and
degrades sanely on ``None``/edge inputs).
"""
from __future__ import annotations
import re
from datetime import UTC, datetime, timedelta
from pathlib import Path
from ci_orchestrator.operations.vms import ParsedClone
from ci_orchestrator.web.filters import (
clone_job,
duration,
humanbytes,
localtime,
relative_time,
shorten_path,
thousands,
)
# ── humanbytes ───────────────────────────────────────────────────────────
def test_humanbytes_example_from_brief() -> None:
assert humanbytes(12345678) == "11.8 MB"
def test_humanbytes_zero() -> None:
assert humanbytes(0) == "0 B"
def test_humanbytes_small_whole_byte_count_has_no_decimal() -> None:
assert humanbytes(500) == "500 B"
def test_humanbytes_kilobytes() -> None:
assert humanbytes(1536) == "1.5 KB"
def test_humanbytes_gigabytes() -> None:
assert humanbytes(1024**3 * 2) == "2.0 GB"
def test_humanbytes_negative_keeps_sign() -> None:
assert humanbytes(-2048) == "-2.0 KB"
def test_humanbytes_accepts_float() -> None:
assert humanbytes(2048.0) == "2.0 KB"
def test_humanbytes_fractional_byte_count_has_decimal() -> None:
assert humanbytes(500.5) == "500.5 B"
# ── duration ─────────────────────────────────────────────────────────────
def test_duration_none_is_dash() -> None:
assert duration(None) == "-"
def test_duration_seconds_only() -> None:
assert duration(45) == "45s"
def test_duration_hours_and_minutes_drops_seconds() -> None:
assert duration(3783) == "1h 03m"
def test_duration_zero() -> None:
assert duration(0) == "0s"
def test_duration_minutes_and_seconds() -> None:
assert duration(125) == "2m 05s"
def test_duration_exactly_one_hour() -> None:
assert duration(3600) == "1h 00m"
def test_duration_negative_keeps_sign() -> None:
assert duration(-45) == "-45s"
def test_duration_accepts_float() -> None:
assert duration(45.9) == "45s"
# ── relative_time ────────────────────────────────────────────────────────
def test_relative_time_none_is_dash() -> None:
assert relative_time(None) == "-"
def test_relative_time_two_minutes_ago() -> None:
ts = (datetime.now(tz=UTC) - timedelta(minutes=2)).isoformat()
assert relative_time(ts) == "2m ago"
def test_relative_time_just_now() -> None:
ts = datetime.now(tz=UTC).isoformat()
assert relative_time(ts) == "just now"
def test_relative_time_hours_ago() -> None:
ts = (datetime.now(tz=UTC) - timedelta(hours=3)).isoformat()
assert relative_time(ts) == "3h ago"
def test_relative_time_days_ago() -> None:
ts = (datetime.now(tz=UTC) - timedelta(days=5)).isoformat()
assert relative_time(ts) == "5d ago"
def test_relative_time_accepts_datetime_object() -> None:
dt = datetime.now(tz=UTC) - timedelta(minutes=10)
assert relative_time(dt) == "10m ago"
def test_relative_time_malformed_string_is_dash() -> None:
assert relative_time("not-a-timestamp") == "-"
def test_relative_time_future_clamps_to_just_now() -> None:
ts = (datetime.now(tz=UTC) + timedelta(minutes=5)).isoformat()
assert relative_time(ts) == "just now"
def test_relative_time_accepts_epoch_float() -> None:
# e.g. a raw os.stat().st_mtime, as stored on inventory.FileEntry.mtime.
epoch = (datetime.now(tz=UTC) - timedelta(minutes=7)).timestamp()
assert relative_time(epoch) == "7m ago"
# ── localtime ────────────────────────────────────────────────────────────
def test_localtime_none_is_dash() -> None:
assert localtime(None) == "-"
_LOCALTIME_SHAPE = re.compile(r"^\d{4}-\d{2}-\d{2} \d{2}:\d{2}$")
def test_localtime_formats_aware_datetime() -> None:
dt = datetime(2026, 7, 23, 10, 0, 0, tzinfo=UTC)
# Exact clock value depends on the host's local zone -- only the shape
# (and that it doesn't raise) is asserted here.
assert _LOCALTIME_SHAPE.match(localtime(dt))
def test_localtime_formats_iso_string() -> None:
assert _LOCALTIME_SHAPE.match(localtime("2026-07-23T10:00:00+00:00"))
def test_localtime_naive_datetime_is_treated_as_already_local() -> None:
dt = datetime(2026, 7, 23, 10, 0, 0) # naive on purpose
assert localtime(dt) == "2026-07-23 10:00"
def test_localtime_malformed_string_is_dash() -> None:
assert localtime("garbage") == "-"
def test_localtime_accepts_epoch_float() -> None:
dt = datetime(2026, 7, 23, 10, 0, 0, tzinfo=UTC)
assert _LOCALTIME_SHAPE.match(localtime(dt.timestamp()))
# ── shorten_path ─────────────────────────────────────────────────────────
def test_shorten_path_default_keep_two() -> None:
assert shorten_path("/var/lib/ci/config.toml") == ".../ci/config.toml"
def test_shorten_path_shorter_than_keep_is_unshortened() -> None:
assert shorten_path("/a") == "/a"
def test_shorten_path_accepts_path_object() -> None:
assert shorten_path(Path("/var/lib/ci/config.toml")) == ".../ci/config.toml"
def test_shorten_path_custom_keep() -> None:
assert shorten_path("/var/lib/ci/logs/web-audit.jsonl", keep=3) == ".../ci/logs/web-audit.jsonl"
def test_shorten_path_keep_zero_is_ellipsis() -> None:
assert shorten_path("/var/lib/ci", keep=0) == "..."
def test_shorten_path_keep_negative_is_ellipsis() -> None:
assert shorten_path("/var/lib/ci", keep=-1) == "..."
# ── thousands ────────────────────────────────────────────────────────────
def test_thousands_small_number_unchanged() -> None:
assert thousands(812) == "812"
def test_thousands_example_from_brief() -> None:
assert thousands(12345) == "12,345"
def test_thousands_zero() -> None:
assert thousands(0) == "0"
def test_thousands_negative() -> None:
assert thousands(-1234) == "-1,234"
def test_thousands_accepts_float() -> None:
assert thousands(12345.0) == "12,345"
# ── clone_job (Stage C1) ─────────────────────────────────────────────────
def test_clone_job_delegates_to_parse_clone_name() -> None:
parsed = clone_job("Clone_build-42_20250101_123456_abc123")
assert parsed == ParsedClone(job_id="build-42", started=datetime(2025, 1, 1, 12, 34, 56))
def test_clone_job_none_input_degrades_to_all_none() -> None:
assert clone_job(None) == ParsedClone(job_id=None, started=None)
def test_clone_job_non_clone_name_degrades_to_all_none() -> None:
assert clone_job("SomeOtherName") == ParsedClone(job_id=None, started=None)
+302
View File
@@ -0,0 +1,302 @@
"""Tests for the Stage B read-only web routes: ``/resources``, ``/vmware``,
``/templates``, ``/files``, plus the B6 backup-form template dropdown.
In-process only, same conventions as ``test_web_app.py``: no real ``/proc``,
``vmrun``, or ``systemctl`` -- every ``operations/`` call is monkeypatched at
the name ``web/app.py`` imported it under.
"""
from __future__ import annotations
import shutil
from pathlib import Path
from typing import Any
import pytest
from fastapi.testclient import TestClient
import ci_orchestrator.web.app as app_module
from ci_orchestrator.config import BackendConfig, Config, Paths
from ci_orchestrator.operations.host import HostResources, VmwareModuleHealth
from ci_orchestrator.operations.inventory import FileEntry, TemplateInfo
from ci_orchestrator.web.app import create_app
def _quiet_systemctl(monkeypatch: pytest.MonkeyPatch) -> None:
"""No real ``systemctl`` subprocess from ``operations.host.runner_status``."""
monkeypatch.setattr(shutil, "which", lambda _name: None)
def _make_config(root: Path) -> Config:
return Config(
paths=Paths(
root=root,
templates=root / "templates",
build_vms=root / "build-vms",
artifacts=root / "artifacts",
keys=root / "keys",
),
backend=BackendConfig(),
)
_RESOURCES = HostResources(
cpu_pct=42.5, ncpu=8, mem_total_gb=16.0, mem_used_gb=8.0,
uptime_sec=3661, load1=0.5, load5=0.4, load15=0.3,
)
_VMWARE_OK = VmwareModuleHealth(
vmmon_loaded=True, vmnet_loaded=True, host_service_active=True, host_service_sub_state="running"
)
_TEMPLATE_ROW = TemplateInfo(
name="LinuxBuild2404", vmx=None, guest_os="ubuntu-64", transport="ssh",
dir_bytes=1024 * 1024, has_vmem=False, last_backup=None,
)
_FILE_ENTRY = FileEntry(name="Template_X_20260101_000000.7z", size_bytes=2048, mtime=1_700_000_000.0)
# ── /resources ───────────────────────────────────────────────────────────────
def test_resources_route_renders_snapshot(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(app_module, "host_resources", lambda: _RESOURCES)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/resources")
assert resp.status_code == 200
assert "Host resources" in resp.text
assert "42.5" in resp.text
def test_resources_route_degrades_on_probe_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom() -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "host_resources", _boom)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/resources")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
# ── /vmware ───────────────────────────────────────────────────────────────────
def test_vmware_route_renders_health(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(app_module, "vmware_module_health", lambda: _VMWARE_OK)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/vmware")
assert resp.status_code == 200
assert "VMware modules" in resp.text
assert "running" in resp.text
def test_vmware_route_degrades_on_probe_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom() -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "vmware_module_health", _boom)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/vmware")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
def test_vmware_route_shows_not_loaded_badge_when_module_missing(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
unhealthy = VmwareModuleHealth(
vmmon_loaded=False, vmnet_loaded=True, host_service_active=False, host_service_sub_state="dead"
)
monkeypatch.setattr(app_module, "vmware_module_health", lambda: unhealthy)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/vmware")
assert resp.status_code == 200
assert "not loaded" in resp.text
assert "down" in resp.text
# ── /templates ────────────────────────────────────────────────────────────────
def test_templates_route_renders_rows(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(app_module, "list_templates", lambda _config: [_TEMPLATE_ROW])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/templates")
assert resp.status_code == 200
assert "LinuxBuild2404" in resp.text
assert "ssh" in resp.text
def test_templates_route_empty_is_not_an_error_notice(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(app_module, "list_templates", lambda _config: [])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/templates")
assert resp.status_code == 200
assert "unavailable" not in resp.text.lower()
assert "No templates found" in resp.text
def test_templates_route_degrades_on_probe_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom(_config: Config) -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "list_templates", _boom)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/templates")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
def test_templates_route_shows_running_ram_snapshot_warning(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
warm = TemplateInfo(
name="Warm", vmx=None, guest_os="ubuntu-64", transport="ssh",
dir_bytes=0, has_vmem=True, last_backup=None,
)
monkeypatch.setattr(app_module, "list_templates", lambda _config: [warm])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/templates")
assert resp.status_code == 200
assert "running-RAM snapshot" in resp.text
# ── /files ────────────────────────────────────────────────────────────────────
def test_files_route_renders_backups_and_artifacts(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(app_module, "list_backups", lambda _config: [_FILE_ENTRY])
monkeypatch.setattr(app_module, "list_artifacts", lambda _config: [])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/files")
assert resp.status_code == 200
assert "Template_X_20260101_000000.7z" in resp.text
assert "No artifacts found" in resp.text
def test_files_route_degrades_on_backups_probe_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom(_config: Config) -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "list_backups", _boom)
monkeypatch.setattr(app_module, "list_artifacts", lambda _config: [])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/files")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
def test_files_route_degrades_on_artifacts_probe_failure(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom(_config: Config) -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "list_backups", lambda _config: [])
monkeypatch.setattr(app_module, "list_artifacts", _boom)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/files")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
# ── dashboard includes all Stage B cards ─────────────────────────────────────
def test_dashboard_renders_all_stage_b_cards(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_quiet_systemctl(monkeypatch)
monkeypatch.setattr(app_module, "host_resources", lambda: _RESOURCES)
monkeypatch.setattr(app_module, "vmware_module_health", lambda: _VMWARE_OK)
monkeypatch.setattr(app_module, "list_templates", lambda _config: [_TEMPLATE_ROW])
monkeypatch.setattr(app_module, "list_backups", lambda _config: [])
monkeypatch.setattr(app_module, "list_artifacts", lambda _config: [])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/")
assert resp.status_code == 200
for needle in ("Host resources", "VMware modules", "Templates", "Template backups", "Artifacts"):
assert needle in resp.text
# ── B6: backup dropdown ──────────────────────────────────────────────────────
def test_host_partial_backup_form_offers_discovered_template_names(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_quiet_systemctl(monkeypatch)
monkeypatch.setattr(app_module, "list_templates", lambda _config: [_TEMPLATE_ROW])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/host")
assert resp.status_code == 200
assert '<select id="host-backup-template" name="template"' in resp.text
assert '<option value="LinuxBuild2404">LinuxBuild2404</option>' in resp.text
assert "7z archive of the template dir under" in resp.text
def test_host_partial_backup_form_empty_dropdown_when_no_templates(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_quiet_systemctl(monkeypatch)
monkeypatch.setattr(app_module, "list_templates", lambda _config: [])
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/host")
assert resp.status_code == 200
assert "<option" not in resp.text
+266
View File
@@ -0,0 +1,266 @@
"""Tests for Stage C: VM-table Job column, Gitea nav link-out, audit polish.
In-process only, same conventions as ``test_web_app.py`` / ``test_web_stage_b.py``:
no real ``vmrun``/disk/systemctl call, everything driven through
``dependency_overrides`` and a fake backend.
"""
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
import pytest
from fastapi.testclient import TestClient
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.config import (
BackendConfig,
CargoCacheConfig,
Config,
Paths,
WebConfig,
WebOidcConfig,
)
from ci_orchestrator.operations.audit import AuditEntry, append_audit
from ci_orchestrator.web.app import create_app, get_backend
def _make_config(
root: Path,
*,
gitea_url: str | None = None,
oidc: WebOidcConfig | None = None,
) -> Config:
return Config(
paths=Paths(
root=root,
templates=root / "templates",
build_vms=root / "build-vms",
artifacts=root / "artifacts",
keys=root / "keys",
),
backend=BackendConfig(),
cargo_cache=CargoCacheConfig(
enabled=True, root=root / "cache" / "cargo", retention_days=30
),
# audit_log must stay under tmp_path -- the real default
# (/var/lib/ci/logs/web-audit.jsonl) is not writable in a test sandbox.
web=WebConfig(audit_log=root / "web-audit.jsonl", gitea_url=gitea_url, oidc=oidc),
)
class _FakeBackend:
"""Stub ``VmBackend`` -- proves no real ``vmrun`` is ever invoked."""
def __init__(self, handles: list[VmHandle], ip: str | None = "192.168.79.201") -> None:
self._handles = handles
self._ip = ip
def list_running(self) -> list[VmHandle]:
return self._handles
def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
return self._ip
def _make_clone_vmx(build_vms: Path, clone_name: str) -> Path:
clone_dir = build_vms / clone_name
clone_dir.mkdir(parents=True)
vmx = clone_dir / f"{clone_name}.vmx"
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
return vmx
# ── VM table Job column (C1) ──────────────────────────────────────────────
def test_vms_table_renders_job_column_header(tmp_path: Path) -> None:
# The Job column header lives inside the VM table, which only renders when
# at least one VM is running (an empty backend shows the "no VMs" notice
# and no table at all), so seed one clone before asserting on the header.
build_vms = tmp_path / "build-vms"
build_vms.mkdir(parents=True)
vmx = _make_clone_vmx(build_vms, "Clone_build-7_20250101_120000_deadbe")
config = _make_config(tmp_path)
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: _FakeBackend([VmHandle(identifier=str(vmx))])
client = TestClient(app)
resp = client.get("/")
assert resp.status_code == 200
assert "<th>Job</th>" in resp.text
def test_vms_table_renders_parsed_job_id_and_started(tmp_path: Path) -> None:
build_vms = tmp_path / "build-vms"
build_vms.mkdir(parents=True)
vmx = _make_clone_vmx(build_vms, "Clone_build-42_20250101_123456_abc123")
config = _make_config(tmp_path)
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: _FakeBackend([VmHandle(identifier=str(vmx))])
client = TestClient(app)
resp = client.get("/vms")
assert resp.status_code == 200
assert "build-42" in resp.text
def test_vms_table_unparseable_name_shows_placeholder_not_error(tmp_path: Path) -> None:
build_vms = tmp_path / "build-vms"
build_vms.mkdir(parents=True)
vmx = _make_clone_vmx(build_vms, "SomeHandCreatedVM")
config = _make_config(tmp_path)
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: _FakeBackend([VmHandle(identifier=str(vmx))])
client = TestClient(app)
resp = client.get("/vms")
assert resp.status_code == 200
assert "—" in resp.text # em-dash placeholder for a missing job_id
# ── Gitea nav link-out (C2) ───────────────────────────────────────────────
def test_dashboard_nav_has_gitea_link_when_configured(tmp_path: Path) -> None:
config = _make_config(tmp_path, gitea_url="https://gitea.example.lan")
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
client = TestClient(app)
resp = client.get("/")
assert resp.status_code == 200
assert 'href="https://gitea.example.lan"' in resp.text
assert 'target="_blank"' in resp.text
assert 'rel="noopener"' in resp.text
def test_dashboard_nav_derives_gitea_link_from_oidc_issuer(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
# Configuring OIDC makes create_app() read the session-secret keyring key
# (absent in the test sandbox); stub the keyring lookup so the app builds
# and we can assert the nav derives its Gitea link from the issuer origin.
monkeypatch.setattr(
"ci_orchestrator.credentials.KeyringCredentialStore.get",
lambda self, name: SimpleNamespace(password="test-session-secret"),
)
oidc = WebOidcConfig(
issuer="https://gitea.emulab.it",
client_id="ci-web",
redirect_url="http://127.0.0.1:8800/auth/callback",
)
config = _make_config(tmp_path, oidc=oidc)
app = create_app(config)
# base.html's nav reads the ``gitea_url`` Jinja global; assert create_app
# installed the issuer-derived origin there. Rendering that global into the
# <a href> is covered end-to-end by test_dashboard_nav_has_gitea_link_when_configured;
# reaching "/" under OIDC would require a full authenticated session this
# wiring check does not need.
assert app.state.templates.env.globals["gitea_url"] == "https://gitea.emulab.it"
def test_dashboard_nav_omits_gitea_link_when_unconfigured(tmp_path: Path) -> None:
config = _make_config(tmp_path) # default WebConfig(): no gitea_url, no oidc
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
client = TestClient(app)
resp = client.get("/")
assert resp.status_code == 200
assert "Gitea" not in resp.text
def test_audit_page_nav_also_gets_gitea_link(tmp_path: Path) -> None:
"""The Jinja global must reach every page that extends base.html, not just '/'."""
config = _make_config(tmp_path, gitea_url="https://gitea.example.lan")
app = create_app(config)
client = TestClient(app)
resp = client.get("/audit")
assert resp.status_code == 200
assert 'href="https://gitea.example.lan"' in resp.text
# ── Audit page polish (C3) ─────────────────────────────────────────────────
def test_audit_page_renders_filter_input(tmp_path: Path) -> None:
config = _make_config(tmp_path)
append_audit(
config.web.audit_log,
AuditEntry(
ts="2026-07-23T10:00:00+00:00",
user="anonymous",
action="vm.remove",
target="/some/vmx",
result="ok",
detail="removed",
),
)
app = create_app(config)
client = TestClient(app)
resp = client.get("/audit")
assert resp.status_code == 200
assert 'id="audit-filter-input"' in resp.text
assert 'aria-controls="audit-rows"' in resp.text
assert '<label for="audit-filter-input">' in resp.text
def test_audit_page_no_filter_input_when_log_empty(tmp_path: Path) -> None:
config = _make_config(tmp_path)
app = create_app(config)
client = TestClient(app)
resp = client.get("/audit")
assert resp.status_code == 200
assert "No audit entries yet." in resp.text
assert 'id="audit-filter-input"' not in resp.text
@pytest.mark.parametrize(
("action", "icon_id"),
[
("vm.remove", "i-server"),
("cache.purge", "i-file"),
("host.backup", "i-box"),
("service.restart", "i-gear"),
("something.else", "i-alert"),
],
)
def test_audit_row_icon_matches_action_prefix(
tmp_path: Path, action: str, icon_id: str
) -> None:
config = _make_config(tmp_path)
append_audit(
config.web.audit_log,
AuditEntry(
ts="2026-07-23T10:00:00+00:00",
user="anonymous",
action=action,
target="target",
result="ok",
detail="detail",
),
)
app = create_app(config)
client = TestClient(app)
resp = client.get("/audit")
assert resp.status_code == 200
assert f'href="#{icon_id}"' in resp.text
+163
View File
@@ -0,0 +1,163 @@
"""Tests for the A4 theming fix.
Root cause being fixed: mutating ``<form method="post">`` do a full-page
navigation, and the POST routes used to return a bare partial with no
``base.html`` wrapper -- an unstyled page. Now the same route renders
inline (bare partial) for an htmx request (``HX-Request: true``) and a
themed full page (``templates/action_page.html``, extending ``base.html``)
for a plain no-JS navigation.
Config/CSRF setup mirrors ``test_web_actions.py``'s ``_make_config`` (a bare,
non-OIDC config is enough -- these routes only need a CSRF-bearing session,
not real authentication).
"""
from __future__ import annotations
import re
from pathlib import Path
from fastapi.testclient import TestClient
from ci_orchestrator.config import BackendConfig, CargoCacheConfig, Config, Paths, WebConfig
from ci_orchestrator.web.app import create_app, get_backend
def _make_config(root: Path) -> Config:
return Config(
paths=Paths(
root=root,
templates=root / "templates",
build_vms=root / "build-vms",
artifacts=root / "artifacts",
keys=root / "keys",
),
backend=BackendConfig(),
cargo_cache=CargoCacheConfig(
enabled=True, root=root / "cache" / "cargo", retention_days=30
),
# audit_log MUST live under tmp_path -- the WebConfig default points
# at the real host's /var/lib/ci/logs, which these tests must never
# touch (see the module docstring: routes exercised here audit).
web=WebConfig(audit_log=root / "web-audit.jsonl"),
)
def _csrf_from(text: str) -> str:
match = re.search(r'name="csrf" value="([^"]+)"', text)
assert match is not None, f"no csrf field found in: {text[:300]!r}"
return match.group(1)
# ── HX-Request present -> bare partial, no theme wrapper ───────────────────
def test_hx_request_preview_gets_bare_partial(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
csrf = _csrf_from(client.get("/cache").text)
resp = client.post(
"/cache/purge", data={"csrf": csrf}, headers={"HX-Request": "true"}
)
assert resp.status_code == 200
assert "<html" not in resp.text.lower()
assert '<link rel="stylesheet"' not in resp.text
assert "Confirm" in resp.text
def test_hx_request_execute_gets_bare_partial(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
csrf = _csrf_from(client.get("/cache").text)
resp = client.post(
"/cache/purge",
data={"csrf": csrf, "stage": "execute"},
headers={"HX-Request": "true"},
)
assert resp.status_code == 200
assert "<html" not in resp.text.lower()
assert '<link rel="stylesheet"' not in resp.text
# ── HX-Request absent -> full themed page ───────────────────────────────────
def test_no_hx_header_preview_gets_full_themed_page(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
csrf = _csrf_from(client.get("/cache").text)
resp = client.post("/cache/purge", data={"csrf": csrf})
assert resp.status_code == 200
assert '<link rel="stylesheet" href="/static/app.css">' in resp.text
assert 'id="action-slot"' in resp.text
assert "Confirm" in resp.text
def test_no_hx_header_execute_gets_full_themed_page(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
csrf = _csrf_from(client.get("/cache").text)
resp = client.post("/cache/purge", data={"csrf": csrf, "stage": "execute"})
assert resp.status_code == 200
assert '<link rel="stylesheet"' in resp.text
assert 'id="action-slot"' in resp.text
# ── wrapper never changes status code, CSRF, or audit behavior ────────────
def test_lock_conflict_status_code_preserved_in_full_page(tmp_path: Path) -> None:
config = _make_config(tmp_path)
config.paths.build_vms.mkdir(parents=True, exist_ok=True)
lock_path = config.paths.build_vms.parent / "vm-start.lock"
lock_path.parent.mkdir(parents=True, exist_ok=True)
lock_path.write_text("job in progress", encoding="utf-8")
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: None
client = TestClient(app)
csrf = _csrf_from(client.get("/cache").text)
resp = client.post("/vms/cleanup", data={"csrf": csrf, "stage": "execute"})
assert resp.status_code == 409
assert '<link rel="stylesheet"' in resp.text
assert 'id="action-slot"' in resp.text
def test_lock_conflict_status_code_preserved_bare_for_htmx(tmp_path: Path) -> None:
config = _make_config(tmp_path)
config.paths.build_vms.mkdir(parents=True, exist_ok=True)
lock_path = config.paths.build_vms.parent / "vm-start.lock"
lock_path.parent.mkdir(parents=True, exist_ok=True)
lock_path.write_text("job in progress", encoding="utf-8")
app = create_app(config)
app.dependency_overrides[get_backend] = lambda: None
client = TestClient(app)
csrf = _csrf_from(client.get("/cache").text)
resp = client.post(
"/vms/cleanup",
data={"csrf": csrf, "stage": "execute"},
headers={"HX-Request": "true"},
)
assert resp.status_code == 409
assert "<html" not in resp.text.lower()
def test_missing_csrf_still_403s_regardless_of_hx_header(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
client.get("/cache") # establish a session
resp = client.post("/cache/purge", data={}, headers={"HX-Request": "true"})
assert resp.status_code == 403