Ali Zarinkolah
80ed5b1577
alembic/env.py previously always overwrote sqlalchemy.url from Settings().postgres.dsn, which made it impossible for a test fixture to point Alembic at a Testcontainers-managed database. Now env.py only sets it when unset, and a new integration suite runs `alembic upgrade head` against a real Postgres container per ADR-0016 (no create_all()).
73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
from collections.abc import AsyncIterator, Iterator
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from alembic.command import upgrade
|
|
from alembic.config import Config
|
|
from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, async_sessionmaker
|
|
from testcontainers.community.postgres import PostgresContainer
|
|
|
|
from src.config import PostgresSettings
|
|
from src.infrastructure.postgres.database import create_engine
|
|
|
|
|
|
def _alembic_config(database_url: str) -> Config:
|
|
config = Config("alembic.ini")
|
|
config.set_main_option("sqlalchemy.url", database_url)
|
|
return config
|
|
|
|
|
|
def _settings_from_url(url: str) -> PostgresSettings:
|
|
# testcontainers returns postgresql+asyncpg://user:pass@host:port/db ;
|
|
# PostgresSettings builds its own dsn from parts, so parse the parts back out.
|
|
without_scheme = url.split("://", 1)[1]
|
|
creds, hostpart = without_scheme.split("@", 1)
|
|
user, password = creds.split(":", 1)
|
|
hostport, db = hostpart.split("/", 1)
|
|
host, port = hostport.split(":", 1)
|
|
return PostgresSettings(host=host, port=int(port), user=user, password=password, db=db)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def postgres_container() -> Iterator[PostgresContainer]:
|
|
with PostgresContainer("postgres:17", driver="asyncpg") as container:
|
|
yield container
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def postgres_url(postgres_container: PostgresContainer) -> str:
|
|
return postgres_container.get_connection_url()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def migrated_postgres_url(postgres_url: str) -> str:
|
|
"""The container's URL, after Alembic has created the schema on it once."""
|
|
upgrade(_alembic_config(postgres_url), "head")
|
|
return postgres_url
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def postgres_engine(migrated_postgres_url: str) -> AsyncIterator[AsyncEngine]:
|
|
engine = create_engine(_settings_from_url(migrated_postgres_url))
|
|
try:
|
|
yield engine
|
|
finally:
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(postgres_engine: AsyncEngine) -> AsyncIterator[AsyncSession]:
|
|
"""One session per test, bound to a rolled-back outer transaction.
|
|
|
|
Isolates each test's writes (ADR-0016: isolate data per test) without
|
|
needing a fresh container or unique keys per test.
|
|
"""
|
|
async with postgres_engine.connect() as connection:
|
|
outer_transaction = await connection.begin()
|
|
sessionmaker = async_sessionmaker(
|
|
bind=connection, expire_on_commit=False, join_transaction_mode="create_savepoint"
|
|
)
|
|
async with sessionmaker() as session:
|
|
yield session
|
|
await outer_transaction.rollback()
|