Simone
9d7bbee35d
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.
245 lines
7.9 KiB
Python
245 lines
7.9 KiB
Python
"""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"
|