diff --git a/libs/db/migrations/versions/0001_initial_schema.py b/libs/db/migrations/versions/0001_initial_schema.py index 9c2b66e..313586a 100644 --- a/libs/db/migrations/versions/0001_initial_schema.py +++ b/libs/db/migrations/versions/0001_initial_schema.py @@ -39,7 +39,7 @@ def upgrade() -> None: sa.Column("issuer_name", sa.Text, nullable=False), sa.Column("exchange", sa.Text, nullable=True), sa.Column("country_code", sa.Text, nullable=True), - sa.Column("is_active", sa.Boolean, nullable=False, server_default="true"), + sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.text("true")), sa.Column( "created_at_utc", sa.DateTime(timezone=True), @@ -69,7 +69,7 @@ def upgrade() -> None: sa.Column("currency", sa.Text, nullable=True), sa.Column("start_date", sa.Date, nullable=True), sa.Column("end_date", sa.Date, nullable=True), - sa.Column("is_primary", sa.Boolean, nullable=False, server_default="true"), + sa.Column("is_primary", sa.Boolean, nullable=False, server_default=sa.text("true")), sa.Column( "created_at_utc", sa.DateTime(timezone=True), @@ -98,12 +98,12 @@ def upgrade() -> None: server_default=sa.text("now()"), ), sa.Column("finished_at_utc", sa.DateTime(timezone=True), nullable=True), - sa.Column("records_seen", sa.Integer, nullable=False, server_default="0"), - sa.Column("records_written", sa.Integer, nullable=False, server_default="0"), - sa.Column("records_skipped", sa.Integer, nullable=False, server_default="0"), - sa.Column("error_count", sa.Integer, nullable=False, server_default="0"), + sa.Column("records_seen", sa.Integer, nullable=False, server_default=sa.text("0")), + sa.Column("records_written", sa.Integer, nullable=False, server_default=sa.text("0")), + sa.Column("records_skipped", sa.Integer, nullable=False, server_default=sa.text("0")), + sa.Column("error_count", sa.Integer, nullable=False, server_default=sa.text("0")), sa.Column("error_summary", sa.Text, nullable=True), - sa.Column("metadata_json", JSONB, nullable=False, server_default="'{}'"), + sa.Column("metadata_json", JSONB, nullable=False, server_default=sa.text("'{}'")), ) op.create_index("ix_job_runs_job_name_run_date", "job_runs", ["job_name", "run_date"]) @@ -128,7 +128,7 @@ def upgrade() -> None: sa.Column("filing_date", sa.Date, nullable=False), sa.Column("accepted_at_utc", sa.DateTime(timezone=True), nullable=True), sa.Column("primary_document_name", sa.Text, nullable=True), - sa.Column("parsed_status", sa.Text, nullable=False, server_default="'pending'"), + sa.Column("parsed_status", sa.Text, nullable=False, server_default=sa.text("'pending'")), sa.Column( "created_at_utc", sa.DateTime(timezone=True), @@ -214,7 +214,7 @@ def upgrade() -> None: sa.Column("filed_at_utc", sa.DateTime(timezone=True), nullable=True), sa.Column("parser_version", sa.Text, nullable=False), sa.Column("parse_confidence", sa.Numeric, nullable=True), - sa.Column("status", sa.Text, nullable=False, server_default="'pending'"), + sa.Column("status", sa.Text, nullable=False, server_default=sa.text("'pending'")), sa.Column( "created_at_utc", sa.DateTime(timezone=True), @@ -260,7 +260,7 @@ def upgrade() -> None: sa.Column("frequency", sa.Text, nullable=True), sa.Column("units", sa.Text, nullable=True), sa.Column("source_name", sa.Text, nullable=False), - sa.Column("metadata_json", JSONB, nullable=False, server_default="'{}'"), + sa.Column("metadata_json", JSONB, nullable=False, server_default=sa.text("'{}'")), sa.Column( "created_at_utc", sa.DateTime(timezone=True), @@ -359,7 +359,7 @@ def upgrade() -> None: sa.Column("stop_price", sa.Numeric, nullable=True), sa.Column("take_profit_price", sa.Numeric, nullable=True), sa.Column("quantity_plan", sa.Numeric, nullable=True), - sa.Column("status", sa.Text, nullable=False, server_default="'draft'"), + sa.Column("status", sa.Text, nullable=False, server_default=sa.text("'draft'")), sa.Column( "created_at_utc", sa.DateTime(timezone=True), @@ -373,7 +373,7 @@ def upgrade() -> None: sa.Column("id", sa.BigInteger, primary_key=True, autoincrement=True), sa.Column("domain", sa.Text, nullable=False), sa.Column("last_sync_at_utc", sa.DateTime(timezone=True), nullable=True), - sa.Column("last_sync_params", JSONB, nullable=False, server_default="'{}'"), + sa.Column("last_sync_params", JSONB, nullable=False, server_default=sa.text("'{}'")), sa.Column("status", sa.Text, nullable=False), sa.Column( "created_at_utc", diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 50da26e..5bfafed 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -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() diff --git a/tests/integration/test_filing_pipeline.py b/tests/integration/test_filing_pipeline.py index e516e7c..1b76570 100644 --- a/tests/integration/test_filing_pipeline.py +++ b/tests/integration/test_filing_pipeline.py @@ -83,6 +83,7 @@ async def test_event_parse_lifecycle(db_session): ticker="AAPL", ) db_session.add(issuer) + await db_session.flush() # Create document doc = Document(