95 lines
3.3 KiB
Python
95 lines
3.3 KiB
Python
|
"""Health surface.
|
||
|
|
|
||
|
|
The point of these tests is the negative cases. An endpoint that returns 200
|
||
|
|
under every condition tells an orchestrator nothing, and the failure is silent
|
||
|
|
exactly when it matters.
|
||
|
|
"""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from pathlib import Path
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from alembic import command
|
||
|
|
from alembic.config import Config
|
||
|
|
from fastapi.testclient import TestClient
|
||
|
|
|
||
|
|
from canned_prompts_service.api import create_app
|
||
|
|
from canned_prompts_service.db import check_readiness, make_engine
|
||
|
|
from canned_prompts_service.settings import Settings
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
||
|
|
|
||
|
|
|
||
|
|
def migrated_url(tmp_path: Path) -> str:
|
||
|
|
url = f"sqlite:///{tmp_path / 'svc.db'}"
|
||
|
|
config = Config(str(ROOT / "alembic.ini"))
|
||
|
|
config.set_main_option("script_location", str(ROOT / "migrations"))
|
||
|
|
config.set_main_option("sqlalchemy.url", url)
|
||
|
|
command.upgrade(config, "head")
|
||
|
|
return url
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture()
|
||
|
|
def ready_client(tmp_path: Path) -> TestClient:
|
||
|
|
url = migrated_url(tmp_path)
|
||
|
|
return TestClient(create_app(Settings(database_url=url), make_engine(url)))
|
||
|
|
|
||
|
|
|
||
|
|
def test_healthz_is_up_without_a_database() -> None:
|
||
|
|
"""Liveness must not depend on the database, or a DB blip restarts pods."""
|
||
|
|
client = TestClient(create_app(Settings(), None))
|
||
|
|
assert client.get("/healthz").json() == {"status": "ok"}
|
||
|
|
|
||
|
|
|
||
|
|
def test_readyz_fails_without_a_database() -> None:
|
||
|
|
client = TestClient(create_app(Settings(), None))
|
||
|
|
response = client.get("/readyz")
|
||
|
|
assert response.status_code == 503
|
||
|
|
assert response.json()["ready"] is False
|
||
|
|
assert "no database" in response.json()["detail"]
|
||
|
|
|
||
|
|
|
||
|
|
def test_readyz_fails_when_the_database_is_unreachable(tmp_path: Path) -> None:
|
||
|
|
engine = make_engine("sqlite:////nonexistent/dir/does-not-exist.db")
|
||
|
|
client = TestClient(create_app(Settings(database_url="x"), engine))
|
||
|
|
response = client.get("/readyz")
|
||
|
|
assert response.status_code == 503
|
||
|
|
assert response.json()["ready"] is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_readyz_fails_when_the_schema_is_not_migrated(tmp_path: Path) -> None:
|
||
|
|
"""Reachable but unmigrated is not ready — it would 500 on the first query."""
|
||
|
|
engine = make_engine(f"sqlite:///{tmp_path / 'empty.db'}")
|
||
|
|
client = TestClient(create_app(Settings(database_url="x"), engine))
|
||
|
|
response = client.get("/readyz")
|
||
|
|
assert response.status_code == 503
|
||
|
|
assert response.json()["detail"] == "schema not migrated"
|
||
|
|
|
||
|
|
|
||
|
|
def test_readyz_reports_the_migration_when_ready(ready_client: TestClient) -> None:
|
||
|
|
body = ready_client.get("/readyz").json()
|
||
|
|
assert body["ready"] is True
|
||
|
|
assert body["migration"] == "0001"
|
||
|
|
|
||
|
|
|
||
|
|
def test_state_health_matches_the_fleet_shape(ready_client: TestClient) -> None:
|
||
|
|
body = ready_client.get("/state/health").json()
|
||
|
|
assert body["status"] == "ok"
|
||
|
|
assert body["service"] == "canned-prompts"
|
||
|
|
assert body["db"] == "connected"
|
||
|
|
assert body["migration"] == "0001"
|
||
|
|
|
||
|
|
|
||
|
|
def test_state_health_degrades_rather_than_lying() -> None:
|
||
|
|
client = TestClient(create_app(Settings(), None))
|
||
|
|
response = client.get("/state/health")
|
||
|
|
assert response.status_code == 503
|
||
|
|
assert response.json()["status"] == "degraded"
|
||
|
|
|
||
|
|
|
||
|
|
def test_settings_have_no_database_fallback() -> None:
|
||
|
|
"""Falling back to a local database when misconfigured hides the mistake."""
|
||
|
|
assert Settings().database_url == ""
|
||
|
|
assert Settings().configured is False
|