You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""Integration test fixtures using testcontainers PostgreSQL."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop_policy():
|
|
return asyncio.DefaultEventLoopPolicy()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def postgres_container():
|
|
"""Start a PostgreSQL container for the test session."""
|
|
try:
|
|
from testcontainers.postgres import PostgresContainer
|
|
|
|
with PostgresContainer("postgres:16-alpine") as pg:
|
|
yield pg
|
|
except ImportError:
|
|
pytest.skip("testcontainers not installed")
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def async_db_url(postgres_container) -> str:
|
|
"""Return asyncpg DSN from the postgres container."""
|
|
sync_url = postgres_container.get_connection_url()
|
|
# Convert psycopg2 URL to asyncpg
|
|
return sync_url.replace("postgresql+psycopg2://", "postgresql+asyncpg://").replace(
|
|
"postgresql://", "postgresql+asyncpg://"
|
|
)
|
|
|
|
|
|
@pytest_asyncio.fixture(scope="session")
|
|
async def db_engine(async_db_url):
|
|
"""Create engine and run migrations."""
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
|
|
# Run migrations via alembic
|
|
alembic_cfg = Config("alembic.ini")
|
|
alembic_cfg.set_main_option("sqlalchemy.url", async_db_url.replace("+asyncpg", "+psycopg2"))
|
|
# Use synchronous URL for alembic
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
engine = create_async_engine(async_db_url, echo=False)
|
|
yield engine
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def db_session(db_engine) -> AsyncSession:
|
|
"""Provide a test DB session that rolls back after each test."""
|
|
factory = async_sessionmaker(db_engine, expire_on_commit=False, class_=AsyncSession)
|
|
async with factory() as session:
|
|
yield session
|
|
await session.rollback()
|