156 lines
4.4 KiB
Python
156 lines
4.4 KiB
Python
import asyncio
|
|
import json
|
|
|
|
import pytest
|
|
|
|
from src import main
|
|
|
|
|
|
def test_health_endpoint_reports_service_liveness_without_gitea_access():
|
|
assert any(getattr(route, "path", None) == "/healthz" for route in main.app.routes)
|
|
assert main.health() == {"status": "ok", "service": "stackchain-dashboard"}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_endpoint_never_contacts_gitea_for_a_request(monkeypatch):
|
|
calls = 0
|
|
|
|
async def unexpected_user_lookup():
|
|
nonlocal calls
|
|
calls += 1
|
|
raise AssertionError("request-time readiness must not contact Gitea")
|
|
|
|
monkeypatch.setattr(main, "_readiness_state", None, raising=False)
|
|
monkeypatch.setattr(main, "current_user", unexpected_user_lookup)
|
|
|
|
first = await main.readiness()
|
|
second = await main.readiness()
|
|
|
|
assert first.status_code == 503
|
|
assert second.status_code == 503
|
|
assert calls == 0
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_monitor_refreshes_on_an_interval_and_cancels(monkeypatch):
|
|
calls = 0
|
|
refreshed_twice = asyncio.Event()
|
|
|
|
async def check():
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 2:
|
|
refreshed_twice.set()
|
|
|
|
monkeypatch.setattr(main, "_check_readiness", check)
|
|
monkeypatch.setattr(main, "READINESS_INTERVAL_SECONDS", 0.01, raising=False)
|
|
|
|
task = asyncio.create_task(main._readiness_monitor())
|
|
await asyncio.wait_for(refreshed_twice.wait(), timeout=0.2)
|
|
task.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await task
|
|
|
|
assert calls == 2
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_endpoint_reports_connected_gitea_user(monkeypatch):
|
|
async def connected_user():
|
|
return {"login": "timmy"}
|
|
|
|
monkeypatch.setattr(main, "current_user", connected_user)
|
|
|
|
await main._check_readiness()
|
|
|
|
response = await main.readiness()
|
|
|
|
assert any(getattr(route, "path", None) == "/readyz" for route in main.app.routes)
|
|
assert response == {
|
|
"status": "ready",
|
|
"service": "stackchain-dashboard",
|
|
"gitea_user": "timmy",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_endpoint_hides_unexpected_upstream_error_details(monkeypatch):
|
|
async def unavailable_user():
|
|
raise RuntimeError("secret internal upstream detail")
|
|
|
|
monkeypatch.setattr(main, "current_user", unavailable_user)
|
|
|
|
await main._check_readiness()
|
|
|
|
response = await main.readiness()
|
|
|
|
assert response.status_code == 503
|
|
assert json.loads(response.body) == {
|
|
"status": "not_ready",
|
|
"service": "stackchain-dashboard",
|
|
"error": "Gitea readiness check is temporarily unavailable",
|
|
}
|
|
assert b"secret internal upstream detail" not in response.body
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_endpoint_returns_503_for_null_current_user_payload(monkeypatch):
|
|
async def null_user():
|
|
return None
|
|
|
|
monkeypatch.setattr(main, "current_user", null_user)
|
|
|
|
await main._check_readiness()
|
|
|
|
response = await main.readiness()
|
|
|
|
assert response.status_code == 503
|
|
assert json.loads(response.body) == {
|
|
"status": "not_ready",
|
|
"service": "stackchain-dashboard",
|
|
"error": "Gitea current-user response did not include a login",
|
|
}
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_endpoint_times_out_and_cancels_stalled_gitea_check(monkeypatch):
|
|
cancelled = asyncio.Event()
|
|
|
|
async def hanging_user():
|
|
try:
|
|
await asyncio.sleep(0.05)
|
|
return {"login": "too-late"}
|
|
finally:
|
|
cancelled.set()
|
|
|
|
monkeypatch.setattr(main, "READINESS_TIMEOUT_SECONDS", 0.01, raising=False)
|
|
monkeypatch.setattr(main, "current_user", hanging_user)
|
|
|
|
await main._check_readiness()
|
|
|
|
response = await main.readiness()
|
|
|
|
assert response.status_code == 503
|
|
assert response.headers["retry-after"] == "1"
|
|
assert json.loads(response.body) == {
|
|
"status": "not_ready",
|
|
"service": "stackchain-dashboard",
|
|
"error": "Gitea readiness check timed out after 0.01s",
|
|
}
|
|
assert cancelled.is_set()
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_readiness_endpoint_expires_a_stale_healthy_snapshot(monkeypatch):
|
|
monkeypatch.setattr(
|
|
main,
|
|
"_readiness_state",
|
|
main.ReadinessState(checked_at=10.0, ready=True, login="timmy"),
|
|
)
|
|
monkeypatch.setattr(main.time, "monotonic", lambda: 76.0)
|
|
|
|
response = await main.readiness()
|
|
|
|
assert response.status_code == 503
|
|
assert json.loads(response.body)["error"] == "Gitea readiness status is not available"
|