feat(api): add CORS middleware and /health contract test
CORS: - New setting CORS_ALLOWED_ORIGINS (comma separated). Defaults cover the three local Vite ports (5173, 5273, 4173); production overlay expects the real origin in .env.prod. - main.py wires CORSMiddleware from settings.cors_origins. No * in production - see RUNBOOK and .env.prod.example. - docker-compose.yml forwards the variable to both api and worker. Tests: - tests/test_api_health.py uses FastAPI TestClient and monkeypatches the five probe functions (postgres/minio/opensearch/qdrant/redis). Verifies the all-ok, any-error, and degraded paths, that the root endpoint reports the configured api prefix, and that the CORS preflight echoes the allowed origin. - pytest tests/test_api_health.py -q: 5 passed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
104
tests/test_api_health.py
Normal file
104
tests/test_api_health.py
Normal file
@@ -0,0 +1,104 @@
|
||||
"""Contract test for /health.
|
||||
|
||||
Patches the individual probe functions so the test does not depend on a live
|
||||
Postgres / MinIO / OpenSearch / Qdrant / Redis. Verifies:
|
||||
|
||||
- the route is mounted under the configured API prefix;
|
||||
- the response shape conforms to ``HealthResponse``;
|
||||
- the overall status follows the worst-component-wins rule
|
||||
(``ok`` -> ``ok``, any ``error`` -> ``error``, ``degraded`` otherwise);
|
||||
- the CORS preflight responds with the configured allowed origin.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from app.api.schemas import ComponentHealth
|
||||
from app.config import settings
|
||||
from app.main import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
def _ok(name: str) -> ComponentHealth:
|
||||
return ComponentHealth(name=name, status="ok", detail={})
|
||||
|
||||
|
||||
def _err(name: str) -> ComponentHealth:
|
||||
return ComponentHealth(name=name, status="error", detail={"error": "down"})
|
||||
|
||||
|
||||
def _degraded(name: str) -> ComponentHealth:
|
||||
return ComponentHealth(name=name, status="degraded", detail={"cluster_status": "red"})
|
||||
|
||||
|
||||
def _patch_probes(monkeypatch, **overrides):
|
||||
from app.api import routes_health
|
||||
|
||||
defaults = {
|
||||
"_check_postgres": lambda: _ok("postgres"),
|
||||
"_check_minio": lambda: _ok("minio"),
|
||||
"_check_opensearch": lambda: _ok("opensearch"),
|
||||
"_check_qdrant": lambda: _ok("qdrant"),
|
||||
"_check_redis": lambda: _ok("redis"),
|
||||
}
|
||||
defaults.update(overrides)
|
||||
for name, fn in defaults.items():
|
||||
monkeypatch.setattr(routes_health, name, fn)
|
||||
|
||||
|
||||
def test_health_all_ok(client: TestClient, monkeypatch):
|
||||
_patch_probes(monkeypatch)
|
||||
res = client.get(f"{settings.app_api_prefix}/health")
|
||||
assert res.status_code == 200
|
||||
body = res.json()
|
||||
assert body["status"] == "ok"
|
||||
assert {c["name"] for c in body["components"]} == {
|
||||
"postgres",
|
||||
"minio",
|
||||
"opensearch",
|
||||
"qdrant",
|
||||
"redis",
|
||||
}
|
||||
|
||||
|
||||
def test_health_error_when_any_component_down(client: TestClient, monkeypatch):
|
||||
_patch_probes(monkeypatch, _check_qdrant=lambda: _err("qdrant"))
|
||||
res = client.get(f"{settings.app_api_prefix}/health")
|
||||
assert res.status_code == 200
|
||||
body = res.json()
|
||||
assert body["status"] == "error"
|
||||
qdrant = next(c for c in body["components"] if c["name"] == "qdrant")
|
||||
assert qdrant["status"] == "error"
|
||||
|
||||
|
||||
def test_health_degraded_when_any_component_degraded(client: TestClient, monkeypatch):
|
||||
_patch_probes(monkeypatch, _check_opensearch=lambda: _degraded("opensearch"))
|
||||
res = client.get(f"{settings.app_api_prefix}/health")
|
||||
body = res.json()
|
||||
assert body["status"] == "degraded"
|
||||
|
||||
|
||||
def test_root_includes_api_prefix(client: TestClient):
|
||||
res = client.get("/")
|
||||
assert res.status_code == 200
|
||||
assert res.json()["api"] == settings.app_api_prefix
|
||||
|
||||
|
||||
def test_cors_preflight_allows_configured_origin(client: TestClient):
|
||||
origin = settings.cors_origins[0]
|
||||
res = client.options(
|
||||
f"{settings.app_api_prefix}/health",
|
||||
headers={
|
||||
"Origin": origin,
|
||||
"Access-Control-Request-Method": "GET",
|
||||
},
|
||||
)
|
||||
assert res.status_code == 200
|
||||
assert res.headers.get("access-control-allow-origin") == origin
|
||||
Reference in New Issue
Block a user