57 lines
1.9 KiB
Python
57 lines
1.9 KiB
Python
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
|
|
from current_status import CurrentStatusCoordinator, build_current_status_router
|
|
|
|
|
|
def app_for(coordinator: CurrentStatusCoordinator) -> FastAPI:
|
|
app = FastAPI()
|
|
app.include_router(build_current_status_router(coordinator))
|
|
return app
|
|
|
|
|
|
async def test_not_found_and_no_check_yet_are_distinct() -> None:
|
|
coordinator = CurrentStatusCoordinator()
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app_for(coordinator)), base_url="http://test"
|
|
) as client:
|
|
missing = await client.get("/targets/missing/status")
|
|
assert missing.status_code == 404
|
|
assert missing.json() == {
|
|
"detail": {"code": "target_not_found", "target_id": "missing"}
|
|
}
|
|
|
|
await coordinator.register_target("new")
|
|
pending = await client.get("/targets/new/status")
|
|
assert pending.status_code == 200
|
|
assert pending.json() == {
|
|
"target_id": "new",
|
|
"state": "not_checked",
|
|
"result": None,
|
|
}
|
|
|
|
|
|
async def test_endpoint_returns_only_latest_result() -> None:
|
|
coordinator = CurrentStatusCoordinator()
|
|
await coordinator.register_target("web")
|
|
|
|
async def first(_target):
|
|
return {"up": False}
|
|
|
|
async def second(_target):
|
|
return {"up": True, "status_code": 204}
|
|
|
|
assert await coordinator.run_check("web", {}, first)
|
|
assert await coordinator.run_check("web", {}, second)
|
|
assert await coordinator.retained_status_count() == 1
|
|
|
|
async with AsyncClient(
|
|
transport=ASGITransport(app=app_for(coordinator)), base_url="http://test"
|
|
) as client:
|
|
response = await client.get("/targets/web/status")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["state"] == "checked"
|
|
assert body["result"]["payload"] == {"up": True, "status_code": 204}
|
|
assert body["result"]["check_sequence"] == 2
|