tegwick
0fb7150956
Found by packaging the service for Railiance (rapp-canned-prompts). Every other rapp in the fleet mounts its database credential as a file; this service could only read CANNED_PROMPTS_DATABASE_URL from the environment, which would have put a database password into kubectl describe, into crash dumps, and in reach of anything able to read /proc. Adds CANNED_PROMPTS_DATABASE_URL_FILE and CANNED_PROMPTS_PUBLISH_TOKEN_FILE. A mounted secret stays a file. When both forms are set the file wins, because a rotated secret must take effect rather than be shadowed by a stale env var, and an unreadable secret file fails loudly rather than falling back to a value that may be older. Service tests 33 -> 36. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01Bjefh8NUiEiahN4JLwoSKM Assistant: claude-code Assistant-Model: opus Assistant-Process: 388925@bnt-lap001 Assistant-Session: 3507023f-e0fd-4a1e-9d90-a0d4217d1502
125 lines
4.7 KiB
Python
125 lines
4.7 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 alembic.script import ScriptDirectory
|
|
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 expected_head() -> str:
|
|
"""Computed, not hardcoded: a new migration must not fail these tests."""
|
|
config = Config(str(ROOT / "alembic.ini"))
|
|
config.set_main_option("script_location", str(ROOT / "migrations"))
|
|
return ScriptDirectory.from_config(config).get_current_head()
|
|
|
|
|
|
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"] == expected_head()
|
|
|
|
|
|
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"] == expected_head()
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_database_url_may_come_from_a_file(tmp_path: Path) -> None:
|
|
"""A mounted secret should stay a file, not become an env var."""
|
|
secret = tmp_path / "url"
|
|
secret.write_text("sqlite:///from-file.db\n", encoding="utf-8")
|
|
settings = Settings(database_url_file=str(secret))
|
|
assert settings.configured is True
|
|
assert settings.resolved_database_url == "sqlite:///from-file.db"
|
|
|
|
|
|
def test_file_wins_over_env_when_both_are_set(tmp_path: Path) -> None:
|
|
"""A rotated secret must take effect, not be shadowed by a stale env var."""
|
|
secret = tmp_path / "url"
|
|
secret.write_text("sqlite:///from-file.db", encoding="utf-8")
|
|
settings = Settings(database_url="sqlite:///from-env.db", database_url_file=str(secret))
|
|
assert settings.resolved_database_url == "sqlite:///from-file.db"
|
|
|
|
|
|
def test_unreadable_secret_file_fails_loudly(tmp_path: Path) -> None:
|
|
settings = Settings(database_url_file=str(tmp_path / "missing"))
|
|
with pytest.raises(RuntimeError, match="cannot read secret file"):
|
|
_ = settings.resolved_database_url
|