Simone
39abfcef80
run.number only counts within its own repo/workflow -- meaningless once the admin endpoint mixes every repo into one list, and already not what the row order is sorted by. run_id is Gitea's actual instance-wide, strictly-increasing id, so the displayed number now agrees with "most recent globally". The link still goes to Gitea's own per-repo run page (html_url); only the label text changed.
1028 lines
34 KiB
Python
1028 lines
34 KiB
Python
"""Tests for the Executions card: route, template, and token handling.
|
|
|
|
Three separate contracts live here:
|
|
|
|
1. ``GET /executions`` renders run rows with their VMs, and degrades to a
|
|
notice + the plain VM rows whenever Gitea cannot be reached.
|
|
2. The Gitea access token from an OIDC login is kept **server-side only** --
|
|
never in the signed-but-unencrypted session cookie (SECURITY-CRITICAL,
|
|
see ``web/auth.py``'s module docstring).
|
|
3. A ``401`` from Gitea triggers exactly one refresh attempt.
|
|
|
|
No network, no keyring, no real ``vmrun``: the backend is injected, the
|
|
keyring is monkeypatched, and Gitea is either stubbed at
|
|
``operations.executions.list_executions`` (template-level tests) or at
|
|
``urlopen`` (one end-to-end pass).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import urllib.error
|
|
from collections import namedtuple
|
|
from datetime import UTC, datetime
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.responses import RedirectResponse
|
|
from fastapi.testclient import TestClient
|
|
|
|
import ci_orchestrator.operations.executions as ex
|
|
import ci_orchestrator.web.app as app_module
|
|
from ci_orchestrator.backends.protocol import VmHandle
|
|
from ci_orchestrator.config import (
|
|
BackendConfig,
|
|
CargoCacheConfig,
|
|
Config,
|
|
Paths,
|
|
WebConfig,
|
|
WebOidcConfig,
|
|
)
|
|
from ci_orchestrator.credentials import Credential, KeyringCredentialStore
|
|
from ci_orchestrator.operations.executions import Execution, ExecutionsView
|
|
from ci_orchestrator.operations.vms import LiveVm
|
|
from ci_orchestrator.web.app import create_app, get_backend
|
|
from ci_orchestrator.web.auth import gitea_token_for, store_gitea_token
|
|
|
|
_SESSION_CRED_NAME = "TestWebSessionSecret"
|
|
_OIDC_CRED_NAME = "TestWebOidcClientSecret"
|
|
_PAT_CRED_NAME = "TestGiteaApiToken"
|
|
_FAKE_PAT = "pat-do-not-leak-123"
|
|
|
|
|
|
# ── fixtures / fakes ────────────────────────────────────────────────────────
|
|
|
|
|
|
def _install_fake_keyring(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
secrets = {
|
|
_SESSION_CRED_NAME: "sess-secret",
|
|
_OIDC_CRED_NAME: "oidc-client-secret",
|
|
_PAT_CRED_NAME: _FAKE_PAT,
|
|
}
|
|
|
|
def fake_get(self: KeyringCredentialStore, target: str) -> Credential:
|
|
if target not in secrets:
|
|
raise KeyError(target)
|
|
return Credential(username=target, password=secrets[target])
|
|
|
|
monkeypatch.setattr(KeyringCredentialStore, "get", fake_get)
|
|
|
|
|
|
def _quiet_host(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_DiskUsage = namedtuple("_DiskUsage", ["total", "used", "free"])
|
|
gb = 1024**3
|
|
monkeypatch.setattr(
|
|
shutil, "disk_usage", lambda _p: _DiskUsage(total=100 * gb, used=40 * gb, free=60 * gb)
|
|
)
|
|
monkeypatch.setattr(shutil, "which", lambda _name: None)
|
|
|
|
|
|
def _make_config(
|
|
root: Path,
|
|
*,
|
|
oidc: bool = False,
|
|
gitea_url: str | None = "https://gitea.example.lan",
|
|
token_cred: str | None = None,
|
|
) -> Config:
|
|
oidc_config = None
|
|
if oidc:
|
|
oidc_config = WebOidcConfig(
|
|
issuer="https://gitea.example.lan",
|
|
client_id="ci-web",
|
|
redirect_url="http://127.0.0.1:8800/auth/callback",
|
|
client_secret_cred=_OIDC_CRED_NAME,
|
|
scopes=["openid", "profile", "read:repository"],
|
|
)
|
|
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
|
|
),
|
|
web=WebConfig(
|
|
enabled=True,
|
|
allowed_users=["Simone"],
|
|
session_secret_cred=_SESSION_CRED_NAME,
|
|
oidc=oidc_config,
|
|
gitea_url=gitea_url,
|
|
gitea_token_cred=token_cred,
|
|
),
|
|
)
|
|
|
|
|
|
class _FakeBackend:
|
|
"""Stub ``VmBackend`` -- proves no real ``vmrun`` is ever invoked."""
|
|
|
|
def __init__(self, handles: list[VmHandle]) -> None:
|
|
self._handles = handles
|
|
|
|
def list_running(self) -> list[VmHandle]:
|
|
return list(self._handles)
|
|
|
|
def get_ip(self, _handle: VmHandle, timeout: float = 0.0) -> str | None:
|
|
return "192.168.79.201"
|
|
|
|
|
|
def _make_clone_vmx(build_vms: Path, name: str) -> Path:
|
|
vmx = build_vms / name / f"{name}.vmx"
|
|
vmx.parent.mkdir(parents=True, exist_ok=True)
|
|
vmx.write_text('guestOS = "ubuntu-64"\n', encoding="utf-8")
|
|
return vmx
|
|
|
|
|
|
def _vm(name: str) -> LiveVm:
|
|
return LiveVm(
|
|
name=name,
|
|
vmx=f"/var/lib/ci/build-vms/{name}/{name}.vmx",
|
|
ip="192.168.79.201",
|
|
guest_os="ubuntu-64",
|
|
uptime_sec=180,
|
|
is_orphan=False,
|
|
)
|
|
|
|
|
|
def _execution(**overrides: Any) -> Execution:
|
|
base: dict[str, Any] = {
|
|
"run_id": 412,
|
|
"number": 12,
|
|
"attempt": 1,
|
|
"workflow": "build-ns7zip.yml",
|
|
"repo": "simone/local-ci-cd-system",
|
|
"branch": "main",
|
|
"commit": "a1b2c3d4e5f60718293a4b5c6d7e8f9012345678",
|
|
"event": "push",
|
|
"actor": "Simone",
|
|
"status": "running",
|
|
"started": datetime(2026, 7, 25, 11, 50, tzinfo=UTC),
|
|
"completed": None,
|
|
"duration_sec": 600,
|
|
"html_url": "https://gitea.example.lan/simone/local-ci-cd-system/actions/runs/12",
|
|
"vms": [],
|
|
}
|
|
base.update(overrides)
|
|
return Execution(**base)
|
|
|
|
|
|
def _stub_view(monkeypatch: pytest.MonkeyPatch, view: ExecutionsView) -> list[str | None]:
|
|
"""Replace the operations call; return the log of tokens it was given."""
|
|
tokens: list[str | None] = []
|
|
|
|
def fake_list_executions(
|
|
_config: Config, _vms: list[LiveVm], *, token: str | None, **_kw: Any
|
|
) -> ExecutionsView:
|
|
tokens.append(token)
|
|
return view
|
|
|
|
monkeypatch.setattr(app_module, "list_executions", fake_list_executions)
|
|
return tokens
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _clear_executions_cache() -> Any:
|
|
ex.clear_cache()
|
|
yield
|
|
ex.clear_cache()
|
|
|
|
|
|
# ── 1. rendering ────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_card_renders_run_row_with_its_vm(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
vm = _vm("Clone_412-1_20260725_115000_linux")
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(vms=[vm])], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "#412" in resp.text
|
|
assert "build-ns7zip.yml" in resp.text
|
|
assert "simone/local-ci-cd-system" in resp.text
|
|
assert "a1b2c3d" in resp.text # commit shortened to 7 chars
|
|
assert "badge-info" in resp.text # running badge
|
|
assert ">running<" in resp.text # never colour-only
|
|
assert vm.name in resp.text
|
|
assert "Remove" in resp.text
|
|
|
|
|
|
def test_attempt_shown_only_when_greater_than_one(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(attempt=2)], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert "#412.2" in resp.text
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "badge"),
|
|
[
|
|
("success", "badge-success"),
|
|
("failure", "badge-danger"),
|
|
("queued", "badge-muted"),
|
|
("cancelled", "badge-muted"),
|
|
("unknown", "badge-muted"),
|
|
],
|
|
)
|
|
def test_status_badges(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, status: str, badge: str
|
|
) -> None:
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(status=status)], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert badge in resp.text
|
|
assert f">{status}<" in resp.text
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("status", "icon"),
|
|
[
|
|
("success", "i-check"),
|
|
("failure", "i-x"),
|
|
],
|
|
)
|
|
def test_status_badge_carries_a_check_or_x_icon(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, status: str, icon: str
|
|
) -> None:
|
|
"""A pass/fail outcome should read without needing the badge colour."""
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(status=status)], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert f'href="#{icon}"' in resp.text
|
|
|
|
|
|
def test_status_badge_carries_a_play_icon_for_running(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""In progress is its own state, not a verdict -- but still not bare text."""
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(status="running")], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
# The card's own header icon is also #i-play -- count, don't just look
|
|
# for presence, or this assertion is trivially true either way.
|
|
assert resp.text.count('href="#i-play"') == 2
|
|
|
|
|
|
@pytest.mark.parametrize("status", ["queued", "cancelled", "unknown"])
|
|
def test_status_badge_has_no_icon_for_non_terminal_states(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, status: str
|
|
) -> None:
|
|
"""queued/cancelled/unknown are neither a verdict nor in progress -- plain text."""
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(status=status)], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert '<td><span class="badge badge-' in resp.text
|
|
assert "#i-check" not in resp.text
|
|
assert "#i-x" not in resp.text
|
|
assert "#i-x" not in resp.text
|
|
|
|
|
|
def test_started_cell_is_a_click_to_toggle_button(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Both the relative and absolute renderings are pre-rendered into the
|
|
button's data attributes -- app.js swaps textContent/title on click, no
|
|
round trip. Server-side, this test can only pin the markup the toggle
|
|
depends on, not the click itself.
|
|
|
|
The id must be stable across the card's own 5s poll (run_id/attempt, not
|
|
anything generated fresh per render) -- app.js keys its client-side
|
|
"which format is showing" memory off it to survive the next poll's
|
|
server-rendered reset back to relative.
|
|
"""
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(run_id=412, attempt=1, started="2026-07-25T11:50:00Z")],
|
|
unmatched_vms=[],
|
|
source="admin",
|
|
error=None,
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert 'id="started-412-1"' in resp.text
|
|
assert 'class="time-toggle text-dim"' in resp.text
|
|
assert 'data-showing="relative"' in resp.text
|
|
assert 'data-relative="' in resp.text
|
|
assert 'data-absolute="' in resp.text
|
|
|
|
|
|
def test_no_cell_stacks_two_facts_on_separate_lines(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Run/Repo, Branch/Commit, Trigger/Actor, Duration/Started and VM/IP each
|
|
get their own column now -- no cell relies on a <br> to show two facts."""
|
|
vm = _vm("Clone_412-1_20260725_115000_linux")
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[_execution(vms=[vm])], unmatched_vms=[], source="admin", error=None
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "<br>" not in resp.text
|
|
for header in ("Repo", "Trigger", "Actor", "Started", "IP / Uptime"):
|
|
assert f"<th>{header}</th>" in resp.text
|
|
assert "Simone" in resp.text # actor, now its own column
|
|
assert vm.ip in resp.text
|
|
|
|
|
|
def test_empty_state_when_no_runs_and_no_vms(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(runs=[], unmatched_vms=[], source="admin", error=None),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert "No recent executions" in resp.text
|
|
assert "<table>" not in resp.text
|
|
|
|
|
|
def test_unmatched_vm_keeps_its_row_and_remove_button(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A powered-on VM with no listed run must not vanish from the dashboard."""
|
|
vm = _vm("Clone_370-1_20260722_101500_linux")
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(runs=[], unmatched_vms=[vm], source="admin", error=None),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert vm.name in resp.text
|
|
assert "370-1" in resp.text # job id parsed out of the clone name
|
|
assert "Remove" in resp.text
|
|
# Same word, same column as a run's "running" status pill -- same icon.
|
|
# The card's own header icon is also #i-play, hence count == 2 not 1.
|
|
assert resp.text.count('href="#i-play"') == 2
|
|
|
|
|
|
def test_end_to_end_against_a_stubbed_gitea(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""No stubbing of the operations layer: only ``urlopen`` is faked."""
|
|
build_vms = tmp_path / "build-vms"
|
|
vmx = _make_clone_vmx(build_vms, "Clone_412-1_20260725_115000_abc")
|
|
payload = {
|
|
"workflow_runs": [
|
|
{
|
|
"id": 412,
|
|
"run_number": 12,
|
|
"run_attempt": 1,
|
|
"status": "in_progress",
|
|
"head_branch": "main",
|
|
"path": ".gitea/workflows/build-ns7zip.yml",
|
|
"repository": {"full_name": "simone/local-ci-cd-system"},
|
|
"html_url": "https://gitea.example.lan/run/12",
|
|
}
|
|
]
|
|
}
|
|
|
|
class _Resp:
|
|
def __enter__(self) -> BytesIO:
|
|
return BytesIO(json.dumps(payload).encode("utf-8"))
|
|
|
|
def __exit__(self, *_exc: object) -> None:
|
|
return None
|
|
|
|
monkeypatch.setattr(ex.urllib.request, "urlopen", lambda _r, timeout=None: _Resp())
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend(
|
|
[VmHandle(identifier=str(vmx))]
|
|
)
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "#412" in resp.text
|
|
assert "build-ns7zip.yml" in resp.text
|
|
# The clone is attached to its run, so it is not listed as "no run".
|
|
assert "no run" not in resp.text
|
|
|
|
|
|
# ── row-count selector ──────────────────────────────────────────────────────
|
|
|
|
|
|
def _stub_limits(monkeypatch: pytest.MonkeyPatch) -> list[int | None]:
|
|
"""Replace the operations call; return the log of limits it was given."""
|
|
limits: list[int | None] = []
|
|
|
|
def fake_list_executions(
|
|
_config: Config,
|
|
_vms: list[LiveVm],
|
|
*,
|
|
token: str | None,
|
|
limit: int | None = None,
|
|
**_kw: Any,
|
|
) -> ExecutionsView:
|
|
limits.append(limit)
|
|
return ExecutionsView(runs=[], unmatched_vms=[], source="admin", error=None)
|
|
|
|
monkeypatch.setattr(app_module, "list_executions", fake_list_executions)
|
|
return limits
|
|
|
|
|
|
def test_default_limit_is_ten(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
limits = _stub_limits(monkeypatch)
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert limits == [10]
|
|
assert '<option value="10" selected>10</option>' in resp.text
|
|
# The selector lives inside the card's header bar, not in a strip below it.
|
|
head = resp.text.split("</h2>", 1)[0]
|
|
assert "head-select" in head
|
|
assert "<label" not in resp.text
|
|
|
|
|
|
def test_selector_offers_the_choices_and_marks_the_current_one(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
limits = _stub_limits(monkeypatch)
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions?limit=50")
|
|
|
|
assert limits == [50]
|
|
for choice in (10, 25, 50, 100):
|
|
assert f'value="{choice}"' in resp.text
|
|
assert '<option value="50" selected>50</option>' in resp.text
|
|
# The re-rendered card polls with the chosen limit, so it survives refreshes.
|
|
assert 'hx-get="/executions?limit=50"' in resp.text
|
|
|
|
|
|
def test_config_default_outside_the_choices_still_appears(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_limits(monkeypatch)
|
|
_install_fake_keyring(monkeypatch)
|
|
config = _make_config(tmp_path, token_cred=_PAT_CRED_NAME)
|
|
object.__setattr__(config.web, "executions_limit", 7)
|
|
app = create_app(config)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert '<option value="7" selected>7</option>' in resp.text
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("query", "expected"),
|
|
[
|
|
("?limit=25", 25),
|
|
("?limit=7", 7), # not offered, but honoured
|
|
("?limit=0", 1), # clamped up
|
|
("?limit=9999", 100), # clamped down
|
|
("?limit=-5", 1),
|
|
("?limit=abc", 10), # unparseable -> config default
|
|
("?limit=", 10),
|
|
("", 10),
|
|
],
|
|
)
|
|
def test_limit_query_is_parsed_permissively(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, query: str, expected: int
|
|
) -> None:
|
|
"""A stale or hand-edited polling URL must never fail the card."""
|
|
limits = _stub_limits(monkeypatch)
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get(f"/executions{query}")
|
|
|
|
assert resp.status_code == 200
|
|
assert limits == [expected]
|
|
|
|
|
|
def test_dashboard_renders_the_selector_at_the_config_default(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_limits(monkeypatch)
|
|
_install_fake_keyring(monkeypatch)
|
|
_quiet_host(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/")
|
|
|
|
assert resp.status_code == 200
|
|
assert 'hx-get="/executions?limit=10"' in resp.text
|
|
|
|
|
|
# ── 2. degraded states ──────────────────────────────────────────────────────
|
|
|
|
|
|
def test_no_token_shows_notice_and_still_lists_vms(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
build_vms = tmp_path / "build-vms"
|
|
vmx = _make_clone_vmx(build_vms, "Clone_412-1_20260725_115000_abc")
|
|
app = create_app(_make_config(tmp_path)) # no token_cred, no login
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend(
|
|
[VmHandle(identifier=str(vmx))]
|
|
)
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "token not configured" in resp.text
|
|
assert "Clone_412-1_20260725_115000_abc" in resp.text
|
|
assert "Remove" in resp.text
|
|
|
|
|
|
def test_gitea_unreachable_degrades(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(
|
|
ex.urllib.request,
|
|
"urlopen",
|
|
lambda _r, timeout=None: (_ for _ in ()).throw(urllib.error.URLError("down")),
|
|
)
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "unreachable" in resp.text
|
|
|
|
|
|
def test_auth_failure_offers_a_relogin_link(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[],
|
|
unmatched_vms=[],
|
|
source="none",
|
|
error="Gitea API returned HTTP 401.",
|
|
auth_failed=True,
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert 'href="/login"' in resp.text
|
|
|
|
|
|
def test_operations_layer_blowing_up_does_not_500(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
def boom(*_args: Any, **_kwargs: Any) -> ExecutionsView:
|
|
raise RuntimeError("unexpected")
|
|
|
|
monkeypatch.setattr(app_module, "list_executions", boom)
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "Executions unavailable" in resp.text
|
|
|
|
|
|
def test_missing_pat_keyring_entry_degrades(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A configured-but-absent keyring PAT must not 500 the card."""
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred="NoSuchCred"))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert "token not configured" in resp.text
|
|
|
|
|
|
def test_pat_is_used_when_configured(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
tokens = _stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(runs=[], unmatched_vms=[], source="admin", error=None),
|
|
)
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
client.get("/executions")
|
|
|
|
assert tokens == [_FAKE_PAT]
|
|
|
|
|
|
def test_old_vms_route_is_gone(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
app = create_app(_make_config(tmp_path))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
assert client.get("/vms").status_code == 404
|
|
|
|
|
|
# ── 3. token handling (SECURITY-CRITICAL) ──────────────────────────────────
|
|
|
|
|
|
class _StubOAuthClient:
|
|
"""Fake Authlib ``gitea`` client -- no network, chosen token/claims."""
|
|
|
|
def __init__(self, token: dict[str, Any], refreshed: dict[str, Any] | None = None) -> None:
|
|
self.token = token
|
|
self.refreshed = refreshed
|
|
self.refresh_calls: list[dict[str, Any]] = []
|
|
|
|
async def authorize_redirect(
|
|
self, request: Any, redirect_uri: str | None = None, **kwargs: Any
|
|
) -> RedirectResponse:
|
|
return RedirectResponse("https://gitea.example.lan/authorize", status_code=302)
|
|
|
|
async def authorize_access_token(self, request: Any, **kwargs: Any) -> dict[str, Any]:
|
|
return self.token
|
|
|
|
async def userinfo(self, **kwargs: Any) -> dict[str, Any]:
|
|
return {"preferred_username": "Simone"}
|
|
|
|
async def fetch_access_token(self, **kwargs: Any) -> dict[str, Any]:
|
|
self.refresh_calls.append(kwargs)
|
|
if self.refreshed is None:
|
|
raise RuntimeError("refresh refused")
|
|
return self.refreshed
|
|
|
|
|
|
def _login(app: Any, client: TestClient) -> None:
|
|
"""Drive the OIDC callback with the stubbed client."""
|
|
resp = client.get("/auth/callback?code=x&state=y", follow_redirects=False)
|
|
assert resp.status_code == 303
|
|
|
|
|
|
def test_login_keeps_token_server_side_and_out_of_the_cookie(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""SECURITY: the session cookie is signed, NOT encrypted -- no token in it."""
|
|
_install_fake_keyring(monkeypatch)
|
|
secret = "gitea-access-token-do-not-leak"
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
app.state.oauth.gitea = _StubOAuthClient(
|
|
{"access_token": secret, "refresh_token": "r3fr3sh", "expires_at": 9999999999}
|
|
)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
_login(app, client)
|
|
|
|
stored = gitea_token_for(app, "Simone")
|
|
assert stored is not None
|
|
assert stored.access_token == secret
|
|
assert stored.refresh_token == "r3fr3sh"
|
|
# Nothing token-shaped anywhere in the cookie jar.
|
|
cookies = "".join(f"{name}={value}" for name, value in client.cookies.items())
|
|
assert secret not in cookies
|
|
assert "r3fr3sh" not in cookies
|
|
|
|
|
|
def test_logout_drops_the_stored_token(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
app.state.oauth.gitea = _StubOAuthClient({"access_token": "tok"})
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
assert gitea_token_for(app, "Simone") is not None
|
|
|
|
client.get("/logout", follow_redirects=False)
|
|
|
|
assert gitea_token_for(app, "Simone") is None
|
|
|
|
|
|
def test_login_token_is_used_for_the_card(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
tokens = _stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(runs=[], unmatched_vms=[], source="admin", error=None),
|
|
)
|
|
_quiet_host(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True, token_cred=_PAT_CRED_NAME))
|
|
app.state.oauth.gitea = _StubOAuthClient({"access_token": "login-token"})
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
|
|
client.get("/executions")
|
|
|
|
# The login token wins over the configured PAT fallback.
|
|
assert tokens == ["login-token"]
|
|
|
|
|
|
def test_scope_refused_login_token_falls_back_to_the_pat(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A logged-in operator whose token lacks API scope must still get runs."""
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True, token_cred=_PAT_CRED_NAME))
|
|
app.state.oauth.gitea = _StubOAuthClient({"access_token": "scope-less"})
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
|
|
seen: list[str | None] = []
|
|
|
|
def fake_list_executions(
|
|
_config: Config, _vms: list[LiveVm], *, token: str | None, **_kw: Any
|
|
) -> ExecutionsView:
|
|
seen.append(token)
|
|
if token == "scope-less":
|
|
return ExecutionsView(
|
|
runs=[],
|
|
unmatched_vms=[],
|
|
source="none",
|
|
error="no API read scope",
|
|
refused=True,
|
|
)
|
|
return ExecutionsView(
|
|
runs=[_execution()], unmatched_vms=[], source="admin", error=None
|
|
)
|
|
|
|
monkeypatch.setattr(app_module, "list_executions", fake_list_executions)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert seen == ["scope-less", _FAKE_PAT]
|
|
assert "#412" in resp.text
|
|
|
|
|
|
def test_no_pat_fallback_when_none_configured(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""Without a PAT there is nothing to retry with -- one call, then degrade."""
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
app.state.oauth.gitea = _StubOAuthClient({"access_token": "scope-less"})
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
tokens = _stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[],
|
|
unmatched_vms=[],
|
|
source="none",
|
|
error="no API read scope",
|
|
refused=True,
|
|
),
|
|
)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert tokens == ["scope-less"]
|
|
assert "no API read scope" in resp.text
|
|
|
|
|
|
def test_pat_is_not_retried_against_itself(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A refused PAT (no login token in play) must not be sent twice."""
|
|
_install_fake_keyring(monkeypatch)
|
|
tokens = _stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[],
|
|
unmatched_vms=[],
|
|
source="none",
|
|
error="no API read scope",
|
|
refused=True,
|
|
),
|
|
)
|
|
app = create_app(_make_config(tmp_path, token_cred=_PAT_CRED_NAME))
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
client.get("/executions")
|
|
|
|
assert tokens == [_FAKE_PAT]
|
|
|
|
|
|
def test_401_triggers_one_refresh_then_retries(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
stub = _StubOAuthClient(
|
|
{"access_token": "stale", "refresh_token": "r3fr3sh"},
|
|
refreshed={"access_token": "fresh"},
|
|
)
|
|
app.state.oauth.gitea = stub
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
|
|
seen: list[str | None] = []
|
|
|
|
def fake_list_executions(
|
|
_config: Config, _vms: list[LiveVm], *, token: str | None, **_kw: Any
|
|
) -> ExecutionsView:
|
|
seen.append(token)
|
|
if token == "stale":
|
|
return ExecutionsView(
|
|
runs=[],
|
|
unmatched_vms=[],
|
|
source="none",
|
|
error="Gitea API returned HTTP 401.",
|
|
auth_failed=True,
|
|
)
|
|
return ExecutionsView(
|
|
runs=[_execution()], unmatched_vms=[], source="admin", error=None
|
|
)
|
|
|
|
monkeypatch.setattr(app_module, "list_executions", fake_list_executions)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert seen == ["stale", "fresh"]
|
|
assert len(stub.refresh_calls) == 1
|
|
assert "#412" in resp.text
|
|
assert gitea_token_for(app, "Simone") is not None
|
|
|
|
|
|
def test_failed_refresh_drops_the_token_and_degrades(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
app.state.oauth.gitea = _StubOAuthClient(
|
|
{"access_token": "stale", "refresh_token": "r3fr3sh"}, refreshed=None
|
|
)
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
_stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(
|
|
runs=[],
|
|
unmatched_vms=[],
|
|
source="none",
|
|
error="Gitea API returned HTTP 401.",
|
|
auth_failed=True,
|
|
),
|
|
)
|
|
|
|
resp = client.get("/executions")
|
|
|
|
assert resp.status_code == 200
|
|
assert 'href="/login"' in resp.text
|
|
assert gitea_token_for(app, "Simone") is None
|
|
|
|
|
|
def test_expired_token_is_refreshed_before_use(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
stub = _StubOAuthClient(
|
|
{"access_token": "old", "refresh_token": "r3fr3sh", "expires_at": 1},
|
|
refreshed={"access_token": "renewed"},
|
|
)
|
|
app.state.oauth.gitea = stub
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
_login(app, client)
|
|
tokens = _stub_view(
|
|
monkeypatch,
|
|
ExecutionsView(runs=[], unmatched_vms=[], source="admin", error=None),
|
|
)
|
|
|
|
client.get("/executions")
|
|
|
|
assert tokens == ["renewed"]
|
|
|
|
|
|
def test_token_without_access_token_is_not_stored(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
"""A login that yields no usable token still logs in; the card degrades."""
|
|
_install_fake_keyring(monkeypatch)
|
|
app = create_app(_make_config(tmp_path, oidc=True))
|
|
app.state.oauth.gitea = _StubOAuthClient({"userinfo": {"preferred_username": "Simone"}})
|
|
app.dependency_overrides[get_backend] = lambda: _FakeBackend([])
|
|
client = TestClient(app)
|
|
|
|
_login(app, client)
|
|
|
|
assert gitea_token_for(app, "Simone") is None
|
|
|
|
|
|
def test_store_and_lookup_helpers_are_per_username(tmp_path: Path) -> None:
|
|
app = create_app(_make_config(tmp_path))
|
|
store_gitea_token(app, "a", {"access_token": "tok-a"})
|
|
store_gitea_token(app, "b", {"access_token": "tok-b"})
|
|
|
|
assert gitea_token_for(app, "a") is not None
|
|
assert gitea_token_for(app, "b") is not None
|
|
assert gitea_token_for(app, "a").access_token == "tok-a" # type: ignore[union-attr]
|
|
assert gitea_token_for(app, None) is None
|
|
assert gitea_token_for(app, "nobody") is None
|