feat(web): read-only dashboard app + ci web serve (W4)

This commit is contained in:
2026-07-23 17:41:06 +02:00
parent 07263e20fb
commit ce37ec2e3b
12 changed files with 791 additions and 0 deletions
+6
View File
@@ -65,6 +65,12 @@ ignore = [
[tool.ruff.lint.per-file-ignores]
"tests/**" = ["B011", "SIM117"]
[tool.ruff.lint.flake8-bugbear]
# FastAPI's `Depends(...)` in a parameter default is the framework's own
# idiom (see web/app.py), not the mutable-default footgun B008 guards
# against -- it is evaluated once and cached by FastAPI's dependency cache.
extend-immutable-calls = ["fastapi.Depends"]
[tool.mypy]
python_version = "3.11"
strict = true
+2
View File
@@ -33,6 +33,7 @@ from ci_orchestrator.commands.wait import ( # noqa: F401 (intentional re-expor
load_backend,
wait_ready,
)
from ci_orchestrator.commands.web import web as web_cmd
@click.group()
@@ -55,6 +56,7 @@ cli.add_command(smoke)
cli.add_command(validate)
cli.add_command(creds)
cli.add_command(vmware)
cli.add_command(web_cmd)
if __name__ == "__main__": # pragma: no cover
+56
View File
@@ -0,0 +1,56 @@
"""``web`` sub-commands -- ``ci web serve`` launches the dashboard app.
**Import discipline (non-negotiable, see CLAUDE.md / the web-UI plan)**: this
module imports only ``click`` and stdlib at module scope. ``fastapi``,
``uvicorn`` and ``ci_orchestrator.web`` (which pulls in fastapi/jinja2) are
imported lazily, inside :func:`serve`'s body -- never at import time. This
keeps ``ci --help`` (and every other CLI command) working on a host where
the optional ``[web]`` extra is not installed.
"""
from __future__ import annotations
import click
@click.group("web")
def web() -> None:
"""Web dashboard commands (read-only in this phase, no auth)."""
@web.command("serve")
@click.option(
"--host",
"host",
default=None,
help="Bind address (default: config web.bind_host, normally 127.0.0.1).",
)
@click.option(
"--port",
"port",
type=int,
default=None,
help="Bind port (default: config web.bind_port).",
)
def serve(host: str | None, port: int | None) -> None:
"""Run the read-only web dashboard in the foreground (uvicorn).
Binds to ``127.0.0.1`` by default (never ``0.0.0.0``) -- pass ``--host``
explicitly (or set ``[web].bind_host`` in ``config.toml``) to expose it
on the LAN.
"""
# Heavy imports deferred to here on purpose -- see module docstring.
import uvicorn
from ci_orchestrator.config import load_config
from ci_orchestrator.web.app import create_app
config = load_config()
bind_host = host if host is not None else config.web.bind_host
bind_port = port if port is not None else config.web.bind_port
app = create_app(config)
uvicorn.run(app, host=bind_host, port=bind_port)
__all__ = ["web"]
+185
View File
@@ -0,0 +1,185 @@
"""FastAPI application factory for the read-only web dashboard (W4).
**In-process only**: routes call ``operations/`` + ``backends.load_backend``
directly and render Jinja2 templates. They never shell out to the CLI (see
``CLAUDE.md`` / the web-UI plan's architecture rules).
**No auth in this step** (that is W5) -- the app binds ``127.0.0.1`` by
default via :class:`~ci_orchestrator.config.WebConfig`; ``create_app`` itself
never chooses the bind address, that is ``commands/web.py``'s job.
**Backend may be unavailable**: :func:`get_backend` is a FastAPI dependency
that swallows :class:`~ci_orchestrator.backends.errors.BackendError` (raised
when ``vmrun`` is missing on this host) and degrades to ``None`` so a route
never 500s. Tests override it via ``app.dependency_overrides[get_backend]``.
"""
from __future__ import annotations
from pathlib import Path
from typing import TYPE_CHECKING
from fastapi import Depends, FastAPI, Request
from fastapi.responses import HTMLResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from ci_orchestrator.backends import load_backend
from ci_orchestrator.backends.errors import BackendError
from ci_orchestrator.operations.cache import CacheStatus, cargo_cache_status
from ci_orchestrator.operations.host import DiskStatus, RunnerStatus, disk_status, runner_status
from ci_orchestrator.operations.vms import LiveVm, list_live_vms
if TYPE_CHECKING: # pragma: no cover
from ci_orchestrator.backends.protocol import VmBackend
from ci_orchestrator.config import Config
_WEB_DIR = Path(__file__).resolve().parent
#: The two real ``act_runner`` systemd units on the Linux host (see
#: CLAUDE.md's "Phase C"). A module constant per the W4 plan section.
RUNNER_SERVICE_NAMES: list[str] = ["act-runner.service", "act-runner-rust.service"]
# ─────────────────────────────────────────────────────────────── dependency
def get_backend(request: Request) -> VmBackend | None:
"""FastAPI dependency: the VM backend for this request, or ``None``.
Catches :class:`BackendError` (e.g. ``vmrun`` missing on this host --
``BackendNotAvailable``) so routes degrade to an empty/"unavailable"
state instead of raising. Overridable in tests via
``app.dependency_overrides[get_backend]`` with a fake backend.
"""
config: Config = request.app.state.config
try:
return load_backend(config)
except BackendError:
return None
# ──────────────────────────────────────────────────────────── data fetchers
def _fetch_live_vms(backend: VmBackend | None, config: Config) -> list[LiveVm]:
"""Render-ready VM list. Degrades to ``[]`` on any backend hiccup."""
if backend is None:
return []
try:
return list_live_vms(backend, config)
except Exception: # a route must never 500 on a backend hiccup
return []
def _fetch_cache_status(config: Config) -> CacheStatus | None:
"""Cargo cache snapshot, or ``None`` if it could not be read."""
try:
return cargo_cache_status(config)
except Exception: # degrade, never 500
return None
def _fetch_disk_status(config: Config) -> DiskStatus | None:
"""Disk usage for ``config.paths.root``, or ``None`` if unreadable."""
try:
return disk_status(config.paths.root)
except Exception: # degrade, never 500
return None
def _fetch_runner_status() -> list[RunnerStatus]:
"""Runner service statuses, or ``[]`` if the query could not run at all."""
try:
return runner_status(RUNNER_SERVICE_NAMES)
except Exception: # degrade, never 500
return []
# ────────────────────────────────────────────────────────────────── routes
def healthz() -> dict[str, str]:
"""Liveness probe for systemd/monitor. Always ``200``."""
return {"status": "ok"}
def dashboard(
request: Request, backend: VmBackend | None = Depends(get_backend)
) -> HTMLResponse:
"""Full dashboard: VM live, cache, disk, runner cards."""
config: Config = request.app.state.config
templates: Jinja2Templates = request.app.state.templates
context = {
"vms": _fetch_live_vms(backend, config),
"backend_available": backend is not None,
"cache": _fetch_cache_status(config),
"disk": _fetch_disk_status(config),
"runners": _fetch_runner_status(),
}
return templates.TemplateResponse(request, "dashboard.html", context)
def vms_partial(
request: Request, backend: VmBackend | None = Depends(get_backend)
) -> HTMLResponse:
"""``GET /vms`` -- polled every 5s by the dashboard via htmx."""
config: Config = request.app.state.config
templates: Jinja2Templates = request.app.state.templates
context = {
"vms": _fetch_live_vms(backend, config),
"backend_available": backend is not None,
}
return templates.TemplateResponse(request, "partials/vms.html", context)
def cache_partial(request: Request) -> HTMLResponse:
"""``GET /cache`` -- cargo cache status partial."""
config: Config = request.app.state.config
templates: Jinja2Templates = request.app.state.templates
context = {"cache": _fetch_cache_status(config)}
return templates.TemplateResponse(request, "partials/cache.html", context)
def host_partial(request: Request) -> HTMLResponse:
"""``GET /host`` -- disk + runner status partial."""
config: Config = request.app.state.config
templates: Jinja2Templates = request.app.state.templates
context = {
"disk": _fetch_disk_status(config),
"runners": _fetch_runner_status(),
}
return templates.TemplateResponse(request, "partials/host.html", context)
# ─────────────────────────────────────────────────────────────────── factory
def create_app(config: Config) -> FastAPI:
"""Build the read-only dashboard app.
W4 scope: no auth, read-only routes only. Binding to a concrete address
(always ``127.0.0.1`` by default -- never ``0.0.0.0``) is
``commands/web.py``'s job, not this factory's.
"""
app = FastAPI(title="ci-orchestrator web")
app.state.config = config
app.state.templates = Jinja2Templates(directory=str(_WEB_DIR / "templates"))
app.mount("/static", StaticFiles(directory=str(_WEB_DIR / "static")), name="static")
app.get("/healthz")(healthz)
app.get("/", response_class=HTMLResponse)(dashboard)
app.get("/vms", response_class=HTMLResponse)(vms_partial)
app.get("/cache", response_class=HTMLResponse)(cache_partial)
app.get("/host", response_class=HTMLResponse)(host_partial)
return app
__all__ = [
"RUNNER_SERVICE_NAMES",
"create_app",
"dashboard",
"get_backend",
"healthz",
]
+1
View File
@@ -0,0 +1 @@
/* placeholder -- real dark Gitea-style design tokens land in W7 */
@@ -0,0 +1,18 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{% block title %}ci-orchestrator{% endblock %}</title>
<link rel="stylesheet" href="/static/app.css">
<script src="/static/htmx.min.js" defer></script>
</head>
<body>
<nav class="navbar">
<span class="navbar-brand">ci-orchestrator</span>
</nav>
<main class="content">
{% block content %}{% endblock %}
</main>
</body>
</html>
@@ -0,0 +1,15 @@
{% extends "base.html" %}
{% block title %}Dashboard - ci-orchestrator{% endblock %}
{% block content %}
<section class="card" id="vms-card">
{% include "partials/vms.html" %}
</section>
<section class="card" id="cache-card">
{% include "partials/cache.html" %}
</section>
{% include "partials/host.html" %}
{% endblock %}
@@ -0,0 +1,15 @@
<div id="cache-partial">
<h2>Cargo cache</h2>
{% if cache %}
<dl>
<dt>Root</dt><dd><code>{{ cache.root }}</code></dd>
<dt>Total size</dt><dd>{{ cache.total_bytes }} bytes</dd>
<dt>Entries</dt><dd>{{ cache.entry_count }}</dd>
<dt>Oldest entry age</dt>
<dd>{{ cache.oldest_age_days ~ " day(s)" if cache.oldest_age_days is not none else "-" }}</dd>
<dt>Retention</dt><dd>{{ cache.retention_days }} day(s)</dd>
</dl>
{% else %}
<p class="notice notice-warn">Cargo cache status unavailable.</p>
{% endif %}
</div>
@@ -0,0 +1,39 @@
<section class="card" id="disk-card">
<h2>Disk</h2>
{% if disk %}
<dl>
<dt>Path</dt><dd><code>{{ disk.path }}</code></dd>
<dt>Total</dt><dd>{{ "%.1f"|format(disk.total_gb) }} GB</dd>
<dt>Free</dt><dd>{{ "%.1f"|format(disk.free_gb) }} GB</dd>
<dt>Used</dt><dd>{{ "%.1f"|format(disk.used_pct) }}%</dd>
</dl>
{% else %}
<p class="notice notice-warn">Disk status unavailable.</p>
{% endif %}
</section>
<section class="card" id="runner-card">
<h2>Runners</h2>
{% if runners %}
<table>
<thead>
<tr>
<th>Service</th>
<th>Active</th>
<th>Sub-state</th>
</tr>
</thead>
<tbody>
{% for runner in runners %}
<tr class="{{ 'row-ok' if runner.active else 'row-alert' }}">
<td>{{ runner.name }}</td>
<td>{{ "yes" if runner.active else "no" }}</td>
<td>{{ runner.sub_state }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p class="notice notice-warn">Runner status unavailable.</p>
{% endif %}
</section>
@@ -0,0 +1,34 @@
<div id="vms-partial" hx-get="/vms" hx-trigger="every 5s" hx-swap="outerHTML">
<h2>Live VMs</h2>
{% if not backend_available %}
<p class="notice notice-warn">Backend unavailable (vmrun not found on this host).</p>
{% endif %}
{% if vms %}
<table>
<thead>
<tr>
<th>Name</th>
<th>VMX</th>
<th>IP</th>
<th>Guest OS</th>
<th>Uptime (s)</th>
<th>Orphan?</th>
</tr>
</thead>
<tbody>
{% for vm in vms %}
<tr class="{{ 'row-orphan' if vm.is_orphan else '' }}">
<td>{{ vm.name }}</td>
<td><code>{{ vm.vmx }}</code></td>
<td>{{ vm.ip or "-" }}</td>
<td>{{ vm.guest_os or "-" }}</td>
<td>{{ vm.uptime_sec if vm.uptime_sec is not none else "-" }}</td>
<td>{{ "yes" if vm.is_orphan else "no" }}</td>
</tr>
{% endfor %}
</tbody>
</table>
{% else %}
<p>No VMs currently running.</p>
{% endif %}
</div>
+150
View File
@@ -0,0 +1,150 @@
"""Tests for ``ci web serve`` (W4 of the web-UI plan).
Two concerns:
1. Behaviour: ``ci web serve --help`` works and ``serve`` builds the app and
calls ``uvicorn.run`` with the right host/port (``uvicorn.run`` is
monkeypatched -- no real server is ever started). ``CI_ROOT``/``CI_CONFIG``
are pointed at an empty tmp dir so the test never depends on this host's
real ``/var/lib/ci/config.toml``.
2. Architecture: ``commands/web.py`` must import only ``click``/stdlib at
module scope. ``fastapi``/``uvicorn``/``ci_orchestrator.web`` may only be
imported lazily, inside the ``serve`` function body -- this keeps
``ci --help`` working when the optional ``[web]`` extra is not installed.
"""
from __future__ import annotations
import ast
import inspect
import sys
from pathlib import Path
from typing import Any
import pytest
from click.testing import CliRunner
import ci_orchestrator.commands.web as web_module
from ci_orchestrator.__main__ import cli
from ci_orchestrator.config import Config
@pytest.fixture(autouse=True)
def _isolated_config_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""Point config resolution at an empty tmp dir (no config.toml present).
Every test in this file exercises ``ci web serve``, which calls
``load_config()``; without this, the real ``/var/lib/ci/config.toml`` on
this host (see CLAUDE.md) could leak in and make assertions flaky.
"""
monkeypatch.setenv("CI_ROOT", str(tmp_path))
monkeypatch.setenv("CI_CONFIG", str(tmp_path / "does-not-exist.toml"))
def _install_fake_uvicorn(monkeypatch: pytest.MonkeyPatch) -> list[dict[str, Any]]:
calls: list[dict[str, Any]] = []
class _FakeUvicorn:
@staticmethod
def run(app: object, **kwargs: Any) -> None:
calls.append({"app": app, **kwargs})
monkeypatch.setitem(sys.modules, "uvicorn", _FakeUvicorn())
return calls
def test_web_serve_help_works() -> None:
runner = CliRunner()
result = runner.invoke(cli, ["web", "serve", "--help"])
assert result.exit_code == 0
assert "--host" in result.output
assert "--port" in result.output
def test_web_serve_calls_uvicorn_run_with_host_and_port(monkeypatch: pytest.MonkeyPatch) -> None:
calls = _install_fake_uvicorn(monkeypatch)
runner = CliRunner()
result = runner.invoke(cli, ["web", "serve", "--host", "127.0.0.1", "--port", "9999"])
assert result.exit_code == 0, result.output
assert len(calls) == 1
assert calls[0]["host"] == "127.0.0.1"
assert calls[0]["port"] == 9999
assert calls[0]["app"] is not None
def test_web_serve_defaults_host_and_port_from_config(monkeypatch: pytest.MonkeyPatch) -> None:
calls = _install_fake_uvicorn(monkeypatch)
runner = CliRunner()
result = runner.invoke(cli, ["web", "serve"])
assert result.exit_code == 0, result.output
assert len(calls) == 1
assert calls[0]["host"] == "127.0.0.1"
assert calls[0]["port"] == 8800
def test_web_serve_host_flag_overrides_config_default_only(
monkeypatch: pytest.MonkeyPatch,
) -> None:
calls = _install_fake_uvicorn(monkeypatch)
runner = CliRunner()
result = runner.invoke(cli, ["web", "serve", "--port", "8123"])
assert result.exit_code == 0, result.output
assert calls[0]["host"] == "127.0.0.1" # config default, untouched
assert calls[0]["port"] == 8123 # flag wins
def test_web_serve_never_binds_0_0_0_0_by_default() -> None:
"""Sanity: the un-overridden default host must never be 0.0.0.0."""
from ci_orchestrator.config import WebConfig
assert WebConfig().bind_host == "127.0.0.1"
# ── architecture: commands/web.py stays dep-light ───────────────────────────
def test_commands_web_module_has_no_top_level_heavy_imports() -> None:
"""Static check: no ``fastapi``/``uvicorn``/``ci_orchestrator.web`` import
at module (top) level -- only inside function bodies. This is what keeps
``ci --help`` working when the optional ``[web]`` extra is not installed.
"""
source = inspect.getsource(web_module)
tree = ast.parse(source)
forbidden_prefixes = ("fastapi", "uvicorn", "ci_orchestrator.web")
for node in tree.body: # module (top) level statements only
if isinstance(node, ast.Import):
for alias in node.names:
assert not alias.name.startswith(
forbidden_prefixes
), f"top-level `import {alias.name}` in commands/web.py"
elif isinstance(node, ast.ImportFrom):
module = node.module or ""
assert not module.startswith(
forbidden_prefixes
), f"top-level `from {module} import ...` in commands/web.py"
def test_commands_web_module_object_carries_no_fastapi_or_uvicorn_attribute() -> None:
"""Belt-and-braces: the module object itself has no fastapi/uvicorn name bound."""
assert not hasattr(web_module, "FastAPI")
assert not hasattr(web_module, "uvicorn")
assert not hasattr(web_module, "create_app")
def test_web_group_registered_on_root_cli() -> None:
assert "web" in cli.commands
assert "serve" in cli.commands["web"].commands # type: ignore[attr-defined]
def test_config_import_does_not_require_web_extra() -> None:
"""``Config``/``load_config`` (imported lazily inside serve) is core, not [web]."""
assert Config is not None
+270
View File
@@ -0,0 +1,270 @@
"""Tests for the read-only (W4) web dashboard app.
In-process only: routes call ``operations/`` + a backend injected via
FastAPI dependency override -- no real ``vmrun`` subprocess is ever
spawned. ``shutil.disk_usage``/``shutil.which`` are monkeypatched so no
real host state (disk, systemd) leaks into the assertions either.
"""
from __future__ import annotations
import shutil
from collections import namedtuple
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.backends.errors import BackendNotAvailable
from ci_orchestrator.backends.protocol import VmHandle
from ci_orchestrator.config import BackendConfig, CargoCacheConfig, Config, Paths
from ci_orchestrator.web.app import create_app, get_backend
_DiskUsage = namedtuple("_DiskUsage", ["total", "used", "free"])
def _fake_disk_usage(_path: Any) -> _DiskUsage:
gb = 1024**3
return _DiskUsage(total=100 * gb, used=40 * gb, free=60 * gb)
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
),
)
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
self.list_running_calls = 0
self.get_ip_calls = 0
def list_running(self) -> list[VmHandle]:
self.list_running_calls += 1
return self._handles
def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
self.get_ip_calls += 1
return self._ip
class _BoomBackend:
"""Backend whose ``list_running`` raises something other than ``BackendError``."""
def list_running(self) -> list[VmHandle]:
raise RuntimeError("boom")
def get_ip(self, handle: VmHandle, timeout: float = 0.0) -> str | None:
return None
def _quiet_host(monkeypatch: pytest.MonkeyPatch) -> None:
"""No real disk/systemctl calls from ``operations.host`` during a test."""
monkeypatch.setattr(shutil, "disk_usage", _fake_disk_usage)
monkeypatch.setattr(shutil, "which", lambda _name: None)
# ── /healthz ──────────────────────────────────────────────────────────────
def test_healthz_returns_ok(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/healthz")
assert resp.status_code == 200
assert resp.json() == {"status": "ok"}
# ── / (dashboard) ────────────────────────────────────────────────────────
def test_dashboard_renders_all_four_cards(
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
for needle in ("Live VMs", "Cargo cache", "Disk", "Runners"):
assert needle in resp.text
def test_dashboard_shows_backend_unavailable_notice_when_backend_is_none(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_quiet_host(monkeypatch)
app = create_app(_make_config(tmp_path))
app.dependency_overrides[get_backend] = lambda: None
client = TestClient(app)
resp = client.get("/")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
# ── /vms ──────────────────────────────────────────────────────────────────
def test_vms_route_renders_injected_vm_and_never_touches_real_vmrun(tmp_path: Path) -> None:
vmx = tmp_path / "build-vms" / "job1" / "job1.vmx"
fake = _FakeBackend([VmHandle(identifier=str(vmx), name="job1")])
app = create_app(_make_config(tmp_path))
app.dependency_overrides[get_backend] = lambda: fake
client = TestClient(app)
resp = client.get("/vms")
assert resp.status_code == 200
assert "job1" in resp.text
assert fake.list_running_calls == 1
def test_vms_route_degrades_when_backend_is_none(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
app.dependency_overrides[get_backend] = lambda: None
client = TestClient(app)
resp = client.get("/vms")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
def test_vms_route_degrades_on_unexpected_backend_error(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
app.dependency_overrides[get_backend] = lambda: _BoomBackend()
client = TestClient(app)
resp = client.get("/vms")
assert resp.status_code == 200
def test_get_backend_degrades_to_none_when_vmrun_missing(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _raise(_config: Config) -> Any:
raise BackendNotAvailable("vmrun not found")
monkeypatch.setattr(app_module, "load_backend", _raise)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/vms")
assert resp.status_code == 200
assert "unavailable" in resp.text.lower()
# ── /cache ────────────────────────────────────────────────────────────────
def test_cache_route_renders_status(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/cache")
assert resp.status_code == 200
assert "Cargo cache" in resp.text
def test_cache_route_degrades_when_operations_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom(_config: Config) -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "cargo_cache_status", _boom)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/cache")
assert resp.status_code == 200
# ── /host ─────────────────────────────────────────────────────────────────
def test_host_route_renders_disk_and_runner_status(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
_quiet_host(monkeypatch)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/host")
assert resp.status_code == 200
assert "Disk" in resp.text
assert "Runners" in resp.text
assert "act-runner.service" in resp.text
assert "act-runner-rust.service" in resp.text
def test_host_route_degrades_when_disk_status_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom(_path: Path) -> Any:
raise RuntimeError("boom")
monkeypatch.setattr(app_module, "disk_status", _boom)
monkeypatch.setattr(shutil, "which", lambda _name: None)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/host")
assert resp.status_code == 200
def test_host_route_degrades_when_runner_status_raises(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
def _boom(_names: list[str]) -> Any:
raise RuntimeError("boom")
_quiet_host(monkeypatch)
monkeypatch.setattr(app_module, "runner_status", _boom)
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/host")
assert resp.status_code == 200
# ── static mount ─────────────────────────────────────────────────────────
def test_static_mount_serves_app_css(tmp_path: Path) -> None:
app = create_app(_make_config(tmp_path))
client = TestClient(app)
resp = client.get("/static/app.css")
assert resp.status_code == 200