fix: integration test isolation and migration server_default escaping
- Fix all JSONB/text server_default values to use sa.text() wrapper to prevent double-escaping in Alembic-generated SQL - Replace testcontainers with direct docker-compose postgres connection in integration conftest, removing asyncio.run() from async fixture context - Change db_engine/db_session to function-scoped with explicit transaction rollback for proper per-test isolation - Flush IssuerMaster before Document insert to respect FK ordering All 94 tests pass. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>main
I Luk Kim
5 months ago
parent
0471018dd8
commit
32a845044c
@ -1,61 +1,29 @@
|
||||
"""Integration test fixtures using testcontainers PostgreSQL."""
|
||||
"""Integration test fixtures using the running postgres from docker-compose."""
|
||||
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
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
||||
|
||||
with PostgresContainer("postgres:16-alpine") as pg:
|
||||
yield pg
|
||||
except ImportError:
|
||||
pytest.skip("testcontainers not installed")
|
||||
# Points to the docker-compose postgres (must be running: docker compose up postgres)
|
||||
TEST_DSN = "postgresql+asyncpg://acef:acef@localhost:5432/acef"
|
||||
|
||||
|
||||
@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)
|
||||
@pytest_asyncio.fixture
|
||||
async def db_engine():
|
||||
"""Function-scoped engine — avoids event loop cross-contamination."""
|
||||
engine = create_async_engine(TEST_DSN, echo=False, pool_pre_ping=True)
|
||||
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()
|
||||
async def db_session(db_engine):
|
||||
"""Session backed by a transaction that rolls back after every test."""
|
||||
async with db_engine.begin() as conn:
|
||||
session = AsyncSession(bind=conn, expire_on_commit=False)
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
await session.close()
|
||||
await conn.rollback()
|
||||
|
||||
Loading…
Reference in New Issue