49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
from sqlalchemy.exc import OperationalError
|
|
|
|
from quartermaster.db import get_session
|
|
from quartermaster.main import create_app
|
|
from fastapi.testclient import TestClient
|
|
|
|
|
|
def test_healthz_returns_ok(client):
|
|
response = client.get("/healthz")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"status": "ok"}
|
|
|
|
|
|
def test_healthz_returns_503_when_db_check_raises(caplog):
|
|
app = create_app()
|
|
|
|
def broken_session():
|
|
class BrokenSession:
|
|
def execute(self, *args, **kwargs):
|
|
raise OperationalError("boom", None, Exception("boom"))
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
session = BrokenSession()
|
|
try:
|
|
yield session
|
|
finally:
|
|
session.close()
|
|
|
|
app.dependency_overrides[get_session] = broken_session
|
|
|
|
caplog.set_level(logging.WARNING, logger="quartermaster.health")
|
|
with TestClient(app) as client:
|
|
response = client.get("/healthz")
|
|
|
|
assert response.status_code == 503
|
|
body = response.json()
|
|
assert body["status"] == "error"
|
|
assert body["detail"] == "OperationalError"
|
|
|
|
failed = [r for r in caplog.records if getattr(r, "event", None) == "healthz_failed"]
|
|
assert len(failed) == 1
|
|
assert failed[0].levelname == "WARNING"
|
|
assert failed[0].error_class == "OperationalError"
|