Simone
8a34e663c7
FastAPI auto-registers these with no auth of its own -- every other data route is individually wrapped in Depends(require_user), but these three are wired up internally by the FastAPI() constructor and were never covered by that. Confirmed live: all three returned 200 with zero session, handing anyone who could reach the port a full map of every route -- VM remove, cache purge, template delete, service control included -- without logging in. Disabled FastAPI's built-in docs_url/redoc_url/openapi_url and re-added all three as ordinary routes behind the same require_user dependency every other route uses. Verified against the live deployment: all three now 303 to /login instead of 200.
340 lines
11 KiB
Python
340 lines
11 KiB
Python
"""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.operations.vms import LiveVm
|
|
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"}
|
|
|
|
|
|
# ── /docs, /redoc, /openapi.json ────────────────────────────────────────
|
|
# Gated behind require_user like every other data route (test_web_auth.py's
|
|
# _PROTECTED_ROUTES proves the redirect-when-unauthenticated half); this
|
|
# proves the positive path still works once past that gate.
|
|
|
|
|
|
def test_openapi_json_lists_a_mutating_route(tmp_path: Path) -> None:
|
|
app = create_app(_make_config(tmp_path))
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/openapi.json")
|
|
|
|
assert resp.status_code == 200
|
|
assert "/vms/{vmx_id}/remove" in resp.json()["paths"]
|
|
|
|
|
|
def test_docs_and_redoc_render(tmp_path: Path) -> None:
|
|
app = create_app(_make_config(tmp_path))
|
|
client = TestClient(app)
|
|
|
|
assert client.get("/docs").status_code == 200
|
|
assert client.get("/redoc").status_code == 200
|
|
|
|
|
|
# ── / (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 ("Executions", "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()
|
|
|
|
|
|
# ── /executions ──────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_executions_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("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "job1" in resp.text
|
|
assert fake.list_running_calls == 1
|
|
|
|
|
|
def test_executions_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("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "unavailable" in resp.text.lower()
|
|
|
|
|
|
def test_executions_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("/executions")
|
|
|
|
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("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "unavailable" in resp.text.lower()
|
|
|
|
|
|
# ── /cache, /runners ──────────────────────────────────────────────────────
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_runners_route_renders_every_unit(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_quiet_host(monkeypatch)
|
|
app = create_app(_make_config(tmp_path))
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/runners")
|
|
|
|
assert resp.status_code == 200
|
|
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:
|
|
"""Stage F put the disk gauge in the Host card, so this is its degrade path."""
|
|
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
|
|
assert "unknown" in resp.text
|
|
|
|
|
|
def test_runners_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("/runners")
|
|
|
|
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
|
|
|
|
|
|
# ── dashboard composition (Stage F) ──────────────────────────────────────
|
|
|
|
_LIVE_VM = LiveVm(
|
|
name="job1", vmx="/x/job1.vmx", ip="10.0.0.5", guest_os="ubuntu-64",
|
|
uptime_sec=120, is_orphan=False,
|
|
)
|
|
|
|
|
|
def test_dashboard_renders_the_info_cards_without_a_status_header(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Stage F: the tiles are gone; every readout they carried is in a card.
|
|
|
|
Threshold levels themselves are covered in ``test_web_cards.py``
|
|
(``_resource_levels``) -- this only pins the page composition.
|
|
"""
|
|
_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"' not in resp.text
|
|
assert 'id="host-card"' in resp.text
|
|
|
|
|
|
def test_dashboard_lists_a_running_vm_under_its_run(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""The VM count left with the tiles -- the Executions card is where VMs show."""
|
|
_quiet_host(monkeypatch)
|
|
vmx = tmp_path / "build-vms" / "job1" / "job1.vmx"
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend(
|
|
[VmHandle(identifier=str(vmx), name=_LIVE_VM.name)]
|
|
)
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/")
|
|
|
|
assert resp.status_code == 200
|
|
assert "job1" in resp.text
|