Luca Sacchi Ricciardi
80 lines
2.6 KiB
Python
80 lines
2.6 KiB
Python
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app import Settings, create_app
|
|
|
|
|
|
def build_settings(db_path: str) -> Settings:
|
|
return Settings(
|
|
db_host="localhost",
|
|
db_port=5432,
|
|
db_name="postgres",
|
|
db_user="user",
|
|
db_password="password",
|
|
ping_query="SELECT 1;",
|
|
ping_interval_minutes=30,
|
|
web_host="0.0.0.0",
|
|
web_port=8080,
|
|
rrd_db_path=db_path,
|
|
rrd_retention_hours=48,
|
|
)
|
|
|
|
|
|
def test_swagger_and_openapi_available():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = str(Path(tmp) / "rrd.sqlite3")
|
|
app = create_app(config_override=build_settings(db_path), start_collector=False)
|
|
|
|
with TestClient(app) as client:
|
|
docs = client.get("/docs")
|
|
assert docs.status_code == 200
|
|
assert "Swagger UI" in docs.text
|
|
|
|
openapi = client.get("/openapi.json")
|
|
assert openapi.status_code == 200
|
|
payload = openapi.json()
|
|
assert "/api/history" in payload["paths"]
|
|
assert "/api/status" in payload["paths"]
|
|
|
|
|
|
def test_history_api_window_and_ring_buffer():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = str(Path(tmp) / "rrd.sqlite3")
|
|
settings = build_settings(db_path)
|
|
app = create_app(config_override=settings, start_collector=False)
|
|
|
|
with TestClient(app) as client:
|
|
store = app.state.store
|
|
now_ts = int(time.time())
|
|
for idx in range(settings.max_samples + 5):
|
|
store.add_sample(
|
|
ts=now_ts - (settings.max_samples + 5) + idx,
|
|
success=(idx % 2 == 0),
|
|
latency_ms=float(idx),
|
|
error_message=None if idx % 2 == 0 else "error",
|
|
)
|
|
|
|
history = client.get("/api/history", params={"hours": 48})
|
|
assert history.status_code == 200
|
|
payload = history.json()
|
|
assert payload["count"] == settings.max_samples
|
|
|
|
invalid = client.get("/api/history", params={"hours": 49})
|
|
assert invalid.status_code == 422
|
|
|
|
|
|
def test_status_api_when_no_samples():
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
db_path = str(Path(tmp) / "rrd.sqlite3")
|
|
app = create_app(config_override=build_settings(db_path), start_collector=False)
|
|
|
|
with TestClient(app) as client:
|
|
status = client.get("/api/status")
|
|
assert status.status_code == 200
|
|
payload = status.json()
|
|
assert payload["success"] is None
|
|
assert payload["timestamp"] is None
|