118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from services.runtime import web, worker
|
|
|
|
|
|
class _Monitor:
|
|
def __init__(self, calls: list[str]) -> None:
|
|
self.calls = calls
|
|
|
|
def probe(self) -> None:
|
|
self.calls.append("monitor")
|
|
|
|
|
|
def test_worker_readiness_reconciles_and_reads_back_public_surface(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
calls: list[str] = []
|
|
monkeypatch.setenv("HUB_OIDC_ISSUER", "https://issuer.example.test/")
|
|
monkeypatch.setattr(
|
|
worker,
|
|
"reconcile_runtime_state",
|
|
lambda client, sessions, catalog_root, issuer: calls.append("runtime"),
|
|
)
|
|
monkeypatch.setattr(
|
|
worker,
|
|
"reconcile_gitea_public_surface",
|
|
lambda client, catalog_root: calls.append("surface") or {"status": "pass"},
|
|
)
|
|
|
|
worker._sync_once(object(), object(), tmp_path, _Monitor(calls))
|
|
|
|
assert calls == ["runtime", "surface", "monitor"]
|
|
|
|
|
|
def test_worker_readiness_fails_closed_when_public_surface_does_not_pass(
|
|
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
|
) -> None:
|
|
monkeypatch.setenv("HUB_OIDC_ISSUER", "https://issuer.example.test/")
|
|
monkeypatch.setattr(worker, "reconcile_runtime_state", lambda *args: None)
|
|
monkeypatch.setattr(
|
|
worker,
|
|
"reconcile_gitea_public_surface",
|
|
lambda *args: {"status": "degraded"},
|
|
)
|
|
|
|
with pytest.raises(RuntimeError, match="public surface reconciliation did not pass"):
|
|
worker._sync_once(object(), object(), tmp_path, _Monitor([]))
|
|
|
|
|
|
def test_worker_error_detail_identifies_parity_failure_without_leaking_exception_text() -> None:
|
|
sensitive_value = "do-not-expose-this-token"
|
|
error = ValueError(
|
|
"ContextForge gateway itworx-opencode tool inventory parity failed "
|
|
f"(missing=1, extra=0, token={sensitive_value})"
|
|
)
|
|
|
|
detail = worker._safe_error_detail(error)
|
|
|
|
assert detail == "runtime-reconciliation:tool-inventory-parity"
|
|
assert sensitive_value not in detail
|
|
|
|
|
|
def test_worker_error_detail_is_bounded_for_unknown_failures() -> None:
|
|
detail = worker._safe_error_detail(RuntimeError("private upstream response"))
|
|
|
|
assert detail == "runtime-reconciliation:RuntimeError"
|
|
assert "private upstream response" not in detail
|
|
|
|
|
|
class _UpstreamResponse:
|
|
def __init__(self, status_code: int) -> None:
|
|
self.status_code = status_code
|
|
|
|
|
|
class _UpstreamClient:
|
|
statuses: dict[str, int] = {}
|
|
|
|
def __init__(self, *, base_url: str, timeout: int) -> None:
|
|
self.base_url = base_url
|
|
|
|
async def get(self, path: str) -> _UpstreamResponse:
|
|
return _UpstreamResponse(self.statuses[f"{self.base_url}{path}"])
|
|
|
|
async def aclose(self) -> None:
|
|
return None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("worker_status", "expected_status"),
|
|
((200, 200), (503, 503)),
|
|
)
|
|
def test_public_readiness_is_gated_by_worker_surface_readback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
tmp_path: Path,
|
|
worker_status: int,
|
|
expected_status: int,
|
|
) -> None:
|
|
(tmp_path / "index.html").write_text("<!doctype html>", encoding="utf-8")
|
|
monkeypatch.setenv("HUB_WEB_ROOT", str(tmp_path))
|
|
monkeypatch.setenv("HUB_CONTROL_URL", "http://control.test")
|
|
monkeypatch.setenv("HUB_WORKER_URL", "http://worker.test")
|
|
_UpstreamClient.statuses = {
|
|
"http://control.test/ready": 200,
|
|
"http://worker.test/health/ready": worker_status,
|
|
}
|
|
monkeypatch.setattr(web.httpx, "AsyncClient", _UpstreamClient)
|
|
|
|
with TestClient(web.build_app()) as client:
|
|
response = client.get("/health/ready")
|
|
|
|
assert response.status_code == expected_status
|
|
assert response.json()["status"] == ("ready" if expected_status == 200 else "not-ready")
|