41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
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():
|
|
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
|
|
|
|
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"
|