56 lines
2.4 KiB
Python
56 lines
2.4 KiB
Python
from datetime import datetime, timezone
|
|
|
|
from app.checker import UnsafeTarget
|
|
from app.models import CheckStatus
|
|
from tests.conftest import create_monitor
|
|
|
|
|
|
class StubChecker:
|
|
def __init__(self, result: CheckStatus | Exception):
|
|
self.result = result
|
|
|
|
async def check(self, url: str) -> CheckStatus:
|
|
if isinstance(self.result, Exception):
|
|
raise self.result
|
|
return self.result
|
|
|
|
|
|
async def test_crud_and_operational_routes(client):
|
|
assert (await client.get("/healthz")).json() == {"status": "ok"}
|
|
assert (await client.get("/readyz")).json() == {"status": "ready"}
|
|
created = await create_monitor(client)
|
|
monitor_id = created["id"]
|
|
assert created["status"]["state"] == "unknown"
|
|
assert len((await client.get("/v1/monitors")).json()) == 1
|
|
assert (await client.get(f"/v1/monitors/{monitor_id}")).json()["name"] == "website"
|
|
updated = await client.patch(f"/v1/monitors/{monitor_id}", json={"name": "api"})
|
|
assert updated.status_code == 200 and updated.json()["name"] == "api"
|
|
assert (await client.patch(f"/v1/monitors/{monitor_id}", json={})).status_code == 422
|
|
assert (await client.delete(f"/v1/monitors/{monitor_id}")).status_code == 204
|
|
missing = await client.get(f"/v1/monitors/{monitor_id}")
|
|
assert missing.status_code == 404
|
|
assert missing.json()["detail"]["code"] == "monitor_not_found"
|
|
|
|
|
|
async def test_check_updates_and_status_route(app, client):
|
|
created = await create_monitor(client)
|
|
result = CheckStatus(state="up", checked_at=datetime.now(timezone.utc),
|
|
status_code=204, latency_ms=12.5)
|
|
app.state.checker = StubChecker(result)
|
|
checked = await client.post(f"/v1/monitors/{created['id']}/checks")
|
|
assert checked.status_code == 200
|
|
assert checked.json()["state"] == "up"
|
|
assert checked.json()["status_applied"] is True
|
|
current = await client.get(f"/v1/monitors/{created['id']}/status")
|
|
assert current.json()["status_code"] == 204
|
|
|
|
|
|
async def test_unsafe_check_maps_error_and_updates_status(app, client):
|
|
created = await create_monitor(client)
|
|
app.state.checker = StubChecker(UnsafeTarget("target resolves to a non-public address"))
|
|
response = await client.post(f"/v1/monitors/{created['id']}/checks")
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"]["code"] == "unsafe_target"
|
|
current = await client.get(f"/v1/monitors/{created['id']}/status")
|
|
assert current.json()["state"] == "error"
|