fix(tests): isolate suite to a dedicated *_test database

Integration fixtures call Base.metadata.drop_all/create_all on get_engine(),
whose DATABASE_URL points at the developer's real DB — localhost:5432/mcma for
host pytest, db:5432/mcma for `make test-api` (pytest runs inside the api
container). Every run silently wiped dev data: drop_all removes ORM tables but
leaves alembic_version (outside Base.metadata), the exact "tables keep
disappearing, version survives" symptom.

conftest now redirects the whole suite to a <db>_test database before settings
load and creates it on demand via asyncpg, so the dev DB is never opened.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Senko-san
2026-06-13 13:27:58 +03:00
parent 0bb752f582
commit 30cb8901f2
+79
View File
@@ -3,10 +3,28 @@
The ASGI app is driven in-process via httpx + asgi-lifespan (no network, no
running server). DB/Redis-backed integration fixtures arrive with the data
layer (plan §11 step 2).
DB safety
---------
Integration fixtures call ``Base.metadata.drop_all`` / ``create_all`` on
``get_engine()``. That engine is built from ``DATABASE_URL``, which in normal
runs points at the *developer's* database — ``localhost:5432/mcma`` for host
``pytest`` and ``db:5432/mcma`` for ``make test-api`` (which execs ``pytest``
inside the api container). Running the suite there silently wipes real data:
``drop_all`` removes every ORM table while leaving Alembic's ``alembic_version``
(it lives outside ``Base.metadata``) — the exact "tables keep disappearing,
version survives" symptom.
To make that impossible, this module redirects every test to a dedicated
``<db>_test`` database *before settings/engine load*, and creates it on demand.
The real dev database is never opened by the test suite.
"""
import asyncio
import os
from collections.abc import AsyncIterator
from pathlib import Path
from urllib.parse import urlsplit, urlunsplit
import pytest
from asgi_lifespan import LifespanManager
@@ -17,6 +35,67 @@ os.environ.setdefault("ENVIRONMENT", "test")
os.environ.setdefault("JWT_SECRET", "test-secret")
def _base_database_url() -> str:
"""Resolve the DB URL the app *would* use, mirroring pydantic-settings
precedence: real env var → ``.env`` file → the app's compiled-in default."""
if env := os.environ.get("DATABASE_URL"):
return env
dotenv = Path(__file__).resolve().parents[1] / ".env"
if dotenv.exists():
for raw in dotenv.read_text().splitlines():
line = raw.strip()
if line.startswith("DATABASE_URL=") and not line.startswith("#"):
return line.split("=", 1)[1].strip().strip("\"'")
return "postgresql+asyncpg://mcma:mcma@localhost:5432/mcma"
def _with_database(url: str, name: str) -> str:
"""Return ``url`` with its database name swapped for ``name``."""
return urlunsplit(urlsplit(url)._replace(path=f"/{name}"))
_BASE_DB_URL = _base_database_url()
_BASE_DB_NAME = urlsplit(_BASE_DB_URL).path.lstrip("/")
# Idempotent: if we're already pointed at a *_test DB, keep it as-is.
_TEST_DB_NAME = _BASE_DB_NAME if _BASE_DB_NAME.endswith("_test") else f"{_BASE_DB_NAME}_test"
_TEST_DB_URL = _with_database(_BASE_DB_URL, _TEST_DB_NAME)
# Redirect the whole suite to the test DB before anything reads settings.
os.environ["DATABASE_URL"] = _TEST_DB_URL
async def _create_test_db_if_missing() -> None:
"""Create ``<db>_test`` if the server is reachable. Best-effort: if Postgres
is down the integration fixtures skip on their own reachability probe, so a
failure here must stay silent rather than break unit-only runs."""
import asyncpg # type: ignore[import-untyped] # driver behind postgresql+asyncpg
# asyncpg wants a plain libpq DSN (no SQLAlchemy "+asyncpg" suffix), against
# the always-present ``postgres`` maintenance database.
dsn = _with_database(_TEST_DB_URL, "postgres").replace("+asyncpg", "")
try:
async with asyncio.timeout(5):
conn = await asyncpg.connect(dsn)
except Exception:
return
try:
exists = await conn.fetchval(
"SELECT 1 FROM pg_database WHERE datname = $1", _TEST_DB_NAME
)
if not exists:
# CREATE DATABASE can't run inside a transaction; asyncpg's implicit
# autocommit on a bare connection handles that.
await conn.execute(f'CREATE DATABASE "{_TEST_DB_NAME}"')
finally:
await conn.close()
@pytest.fixture(scope="session", autouse=True)
def _ensure_test_database() -> None:
"""Guarantee the dedicated test database exists once per session."""
asyncio.run(_create_test_db_if_missing())
@pytest.fixture
async def client() -> AsyncIterator[AsyncClient]:
from app.main import create_app