feat(web): Executions card replaces Live VMs
The dashboard could only show what was powered on this instant, so it could
not answer "did the last build pass?" — and no local execution history exists
to answer it from (`report job` reads a PowerShell-era JSONL path that nothing
writes on this host, and the recent artifact dirs are empty).
Recent runs now come from Gitea's Actions API (`operations/executions.py`):
`/api/v1/admin/actions/runs` for every repo on the instance, falling back to
`/api/v1/user/actions/runs` when the token is not admin-scoped. One list call
per refresh, cached 10s, so the card's 5s htmx poll costs Gitea at most one
call per window. Live VMs are grouped under the run that spawned them — the
job id is `{run_id}-{run_attempt}[-suffix]` and the clone name carries it — so
a matrix run collects several VMs on one row. Every failure mode (no token,
401, unreachable Gitea, malformed payload) degrades to a notice plus the plain
VM rows, each keeping its Remove button, so the card is never less useful than
the table it replaces. Orphans stay where they were: the sweep in Maintenance,
the count in the status-header tile.
Auth reuses the access token from the operator's own OIDC login, which
`auth.py` previously discarded. SECURITY: that token is kept server-side only
(`app.state.gitea_tokens`, dropped on logout) and never written into the
session — `SessionMiddleware` signs the session cookie but does not encrypt
it, and this service is deployed over plain HTTP on the LAN, so a bearer token
in there would be readable in the browser and on the wire. An expired token is
refreshed once; `[web] gitea_token_cred` names an optional keyring PAT for the
cases a login token cannot cover (service restart, non-admin operator).
`GET /vms` and `partials/vms.html` are gone; `POST /vms/{id}/remove` and
`POST /vms/cleanup` are untouched.
Design: docs/superpowers/specs/2026-07-25-executions-card-design.md
This commit is contained in:
@@ -0,0 +1,686 @@
|
||||
"""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 "#12" 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 "#12.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
|
||||
|
||||
|
||||
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
|
||||
|
||||
|
||||
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 "#12" 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
|
||||
|
||||
|
||||
# ── 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_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 "#12" 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
|
||||
Reference in New Issue
Block a user