59 lines
2.8 KiB
Python
59 lines
2.8 KiB
Python
import httpx
|
|
import pytest
|
|
|
|
from app.main import app
|
|
from tests.conftest import install_checker
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
async def test_crud_and_status_lifecycle(api_client):
|
|
created = await api_client.post("/api/v1/monitors", json={"name": "Example", "url": "https://example.com/path?secret=yes"})
|
|
assert created.status_code == 201
|
|
monitor_id = created.json()["id"]
|
|
assert created.json()["status"] == "unknown"
|
|
assert (await api_client.get("/api/v1/monitors")).json()[0]["id"] == monitor_id
|
|
|
|
updated = await api_client.put(f"/api/v1/monitors/{monitor_id}", json={"name": "New", "url": "https://example.org"})
|
|
assert updated.status_code == 200
|
|
assert updated.json()["name"] == "New"
|
|
assert (await api_client.get(f"/api/v1/monitors/{monitor_id}/status")).json()["status"] == "unknown"
|
|
assert (await api_client.delete(f"/api/v1/monitors/{monitor_id}")).status_code == 204
|
|
assert (await api_client.get(f"/api/v1/monitors/{monitor_id}")).status_code == 404
|
|
|
|
|
|
async def test_validation_and_probes(api_client):
|
|
assert (await api_client.post("/api/v1/monitors", json={"name": "", "url": "file:///tmp/x"})).status_code == 422
|
|
assert (await api_client.get("/healthz")).json() == {"status": "ok"}
|
|
assert (await api_client.get("/readyz")).json() == {"status": "ready"}
|
|
|
|
|
|
async def test_check_updates_status(api_client):
|
|
install_checker(lambda request: httpx.Response(204, request=request))
|
|
item = (await api_client.post("/api/v1/monitors", json={"name": "ok", "url": "https://example.com"})).json()
|
|
checked = await api_client.post(f"/api/v1/monitors/{item['id']}/check")
|
|
assert checked.status_code == 200
|
|
assert checked.json()["status"] == "up"
|
|
assert checked.json()["http_status"] == 204
|
|
assert checked.json()["latency_ms"] >= 0
|
|
assert (await api_client.get(f"/api/v1/monitors/{item['id']}/status")).json()["status"] == "up"
|
|
|
|
|
|
async def test_http_failure_is_down(api_client):
|
|
install_checker(lambda request: httpx.Response(503, request=request))
|
|
item = (await api_client.post("/api/v1/monitors", json={"name": "bad", "url": "https://example.com"})).json()
|
|
checked = await api_client.post(f"/api/v1/monitors/{item['id']}/check")
|
|
assert checked.json()["status"] == "down"
|
|
assert checked.json()["http_status"] == 503
|
|
|
|
|
|
async def test_timeout_is_sanitized_error(api_client):
|
|
def timeout(request):
|
|
raise httpx.ReadTimeout("contains-a-secret", request=request)
|
|
install_checker(timeout)
|
|
item = (await api_client.post("/api/v1/monitors", json={"name": "slow", "url": "https://example.com"})).json()
|
|
checked = await api_client.post(f"/api/v1/monitors/{item['id']}/check")
|
|
assert checked.json()["status"] == "error"
|
|
assert checked.json()["error"] == "timeout"
|
|
assert "contains-a-secret" not in checked.text
|