67 lines
2.4 KiB
Python
67 lines
2.4 KiB
Python
"""Shared fixtures for tests that need a real PostgreSQL.
|
|
|
|
In docker compose TEST_DATABASE_URL is already set (database `<name>_test`).
|
|
Elsewhere set it to a throwaway database whose name ends with `_test`; it is
|
|
created if missing and its tables are emptied by every test that uses `session`.
|
|
Without TEST_DATABASE_URL those tests are skipped.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import text
|
|
from sqlalchemy.engine import make_url
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from app.config import settings
|
|
from app.core.database.connection import create_engine, create_session_factory
|
|
from app.core.database.repository import TrainingRepository
|
|
|
|
TEST_DATABASE_URL = os.environ.get("TEST_DATABASE_URL")
|
|
|
|
|
|
async def ensure_test_database(url: str) -> None:
|
|
"""Create the test database if it doesn't exist yet."""
|
|
target = make_url(url)
|
|
name = target.database or ""
|
|
if not name.endswith("_test"):
|
|
pytest.fail(f"Refusing to run: database {name!r} must end with '_test'")
|
|
|
|
admin_engine = create_async_engine(
|
|
target.set(database="postgres"), isolation_level="AUTOCOMMIT"
|
|
)
|
|
try:
|
|
async with admin_engine.connect() as conn:
|
|
exists = await conn.scalar(
|
|
text("SELECT 1 FROM pg_database WHERE datname = :name"), {"name": name}
|
|
)
|
|
if not exists:
|
|
quoted = name.replace('"', '""')
|
|
await conn.execute(text(f'CREATE DATABASE "{quoted}"'))
|
|
finally:
|
|
await admin_engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def alembic_config(monkeypatch) -> Config:
|
|
if not TEST_DATABASE_URL:
|
|
pytest.skip("TEST_DATABASE_URL is not set")
|
|
await ensure_test_database(TEST_DATABASE_URL)
|
|
# migrations/env.py takes the URL from settings
|
|
monkeypatch.setattr(settings, "POSTGRES_DATABASE_URL", TEST_DATABASE_URL)
|
|
return Config("alembic.ini")
|
|
|
|
|
|
@pytest.fixture
|
|
async def session(alembic_config):
|
|
"""Session on a migrated and emptied test database."""
|
|
# env.py runs its own event loop, so alembic goes to a thread
|
|
await asyncio.to_thread(command.upgrade, alembic_config, "head")
|
|
engine = create_engine(TEST_DATABASE_URL)
|
|
async with create_session_factory(engine)() as session:
|
|
await TrainingRepository(session).delete_all()
|
|
yield session
|
|
await engine.dispose()
|