0d898e09f2
run_all_migrations() now adopts/initializes only the app DB and returns
{'app': ...}. app/main.py drops the location/poo readiness checks
(ensure_location_db_ready / ensure_poo_db_ready) and their imports;
ensure_runtime_dirs only provisions the app DB path; lifespan still
fail-closes on a missing/unmanaged app DB. Delete the retired
location/poo adopt scripts and the alembic_location / alembic_poo
chains. Update tests to single-DB expectations and drop the obsolete
location/poo adoption + readiness tests.
pytest 95 passed; ruff clean (pre-existing only); a fresh app DB
initialized via scripts.run_migrations contains location + poo_records.
112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
from pathlib import Path
|
|
import sqlite3
|
|
|
|
import anyio
|
|
import pytest
|
|
import yaml
|
|
|
|
from app.db import reset_db_caches
|
|
from app.config import get_settings
|
|
from app.main import create_app
|
|
from scripts.app_db_adopt import APP_BASELINE_REVISION
|
|
from scripts.run_migrations import run_all_migrations
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _read_yaml(path: str) -> dict:
|
|
return yaml.safe_load((PROJECT_ROOT / path).read_text())
|
|
|
|
|
|
async def _run_lifespan(app) -> None:
|
|
async with app.router.lifespan_context(app):
|
|
return None
|
|
|
|
|
|
def _configure_database_env(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> dict[str, Path | str]:
|
|
app_path = tmp_path / "app.db"
|
|
|
|
monkeypatch.setenv("APP_DATABASE_URL", f"sqlite:///{app_path}")
|
|
monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
|
|
monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
|
|
monkeypatch.setenv("AUTH_COOKIE_SECURE_OVERRIDE", "false")
|
|
get_settings.cache_clear()
|
|
reset_db_caches()
|
|
|
|
return {
|
|
"app_path": app_path,
|
|
"app_url": f"sqlite:///{app_path}",
|
|
}
|
|
|
|
|
|
def test_compose_uses_migration_job_before_app() -> None:
|
|
compose = _read_yaml("docker-compose.yml")
|
|
override = _read_yaml("docker-compose.override.yml")
|
|
|
|
migration_service = compose["services"]["migration"]
|
|
app_service = compose["services"]["app"]
|
|
|
|
assert migration_service["command"] == ["python", "-m", "scripts.run_migrations"]
|
|
assert migration_service["restart"] == "no"
|
|
assert app_service["depends_on"]["migration"]["condition"] == "service_completed_successfully"
|
|
assert override["services"]["migration"]["build"] == "."
|
|
assert override["services"]["app"]["build"] == "."
|
|
|
|
|
|
def test_image_defaults_to_uvicorn_only() -> None:
|
|
dockerfile = (PROJECT_ROOT / "Dockerfile").read_text()
|
|
entrypoint = (PROJECT_ROOT / "docker/entrypoint.sh").read_text()
|
|
|
|
assert 'CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]' in dockerfile
|
|
assert 'exec "$@"' in entrypoint
|
|
assert "app_db_adopt" not in entrypoint
|
|
assert "location_db_adopt" not in entrypoint
|
|
assert "poo_db_adopt" not in entrypoint
|
|
|
|
|
|
def test_migration_runner_initializes_and_is_idempotent(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
database_urls = _configure_database_env(tmp_path, monkeypatch)
|
|
|
|
first_run = run_all_migrations()
|
|
second_run = run_all_migrations()
|
|
|
|
assert first_run == {"app": "initialized"}
|
|
assert second_run == {"app": "already_managed"}
|
|
|
|
conn = sqlite3.connect(database_urls["app_path"])
|
|
try:
|
|
assert conn.execute("SELECT version_num FROM alembic_version").fetchone()[0] == APP_BASELINE_REVISION
|
|
tables = {
|
|
row[0]
|
|
for row in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'"
|
|
).fetchall()
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
assert {
|
|
"auth_users", "auth_sessions", "app_config", "alembic_version", "location", "poo_records"
|
|
} <= tables
|
|
|
|
get_settings.cache_clear()
|
|
reset_db_caches()
|
|
|
|
|
|
def test_app_startup_still_fails_closed_without_running_adoption(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
database_urls = _configure_database_env(tmp_path, monkeypatch)
|
|
missing_app_path = database_urls["app_path"]
|
|
|
|
app = create_app()
|
|
with pytest.raises(RuntimeError, match="Run 'python scripts/app_db_adopt.py' first"):
|
|
anyio.run(_run_lifespan, app)
|
|
|
|
assert not Path(missing_app_path).exists()
|
|
|
|
get_settings.cache_clear()
|
|
reset_db_caches()
|