80 lines
2.5 KiB
Python
80 lines
2.5 KiB
Python
"""``api/system.py`` 健康探针端点。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import types
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from kilostar.api.system import system_router
|
|
|
|
|
|
@pytest.fixture
|
|
def health_app() -> FastAPI:
|
|
app = FastAPI()
|
|
app.include_router(system_router)
|
|
return app
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_liveness_returns_alive(health_app):
|
|
transport = ASGITransport(app=health_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/health/live")
|
|
assert resp.status_code == 200
|
|
assert resp.json() == {"status": "alive"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_readiness_all_ok(health_app, fake_actors):
|
|
pg = types.SimpleNamespace()
|
|
pg.ping = AsyncMock(return_value=True)
|
|
fake_actors.register("postgres_database", pg)
|
|
|
|
gsm = types.SimpleNamespace()
|
|
gsm.get_skill_list = AsyncMock(return_value=[])
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
transport = ASGITransport(app=health_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/health/ready")
|
|
|
|
assert resp.status_code == 200
|
|
body = resp.json()
|
|
assert body["status"] == "ready"
|
|
assert body["checks"] == {"postgres": True, "global_state_machine": True}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_readiness_postgres_down(health_app, fake_actors):
|
|
pg = types.SimpleNamespace()
|
|
pg.ping = AsyncMock(side_effect=RuntimeError("db down"))
|
|
fake_actors.register("postgres_database", pg)
|
|
|
|
gsm = types.SimpleNamespace()
|
|
gsm.get_skill_list = AsyncMock(return_value=[])
|
|
fake_actors.register("global_state_machine", gsm)
|
|
|
|
transport = ASGITransport(app=health_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/health/ready")
|
|
|
|
assert resp.status_code == 503
|
|
body = resp.json()
|
|
assert body["status"] == "not_ready"
|
|
assert body["checks"]["postgres"] is False
|
|
assert body["checks"]["global_state_machine"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_readiness_actor_not_registered(health_app, fake_actors):
|
|
transport = ASGITransport(app=health_app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
resp = await client.get("/health/ready")
|
|
|
|
assert resp.status_code == 503
|
|
assert resp.json()["status"] == "not_ready"
|