FabledCurator/tests/conftest.py

"""Shared pytest fixtures.

The async db fixture provides an AsyncSession bound to a transaction that
gets rolled back after each test. CI provisions a real Postgres + pgvector
(see .forgejo/workflows/ci.yml), so tests exercise the actual schema and
migration code paths.
"""

import os

# Audit 2026-06-02: CredentialCrypto now refuses to auto-generate a
# Fernet key without explicit opt-in (production safety against silent
# key regeneration on partial restore). The test environment never has
# a pre-seeded key file, so set the bootstrap flag here before any
# create_app() / CredentialCrypto() import path fires.
os.environ.setdefault("CURATOR_BOOTSTRAP_NEW_KEY", "1")

import pytest  # noqa: E402
import pytest_asyncio  # noqa: E402
from sqlalchemy import create_engine  # noqa: E402
from sqlalchemy.ext.asyncio import (  # noqa: E402
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import sessionmaker  # noqa: E402

from backend.app import create_app  # noqa: E402
from backend.app.models import Base  # noqa: E402


def _async_database_url() -> str:
    user = os.environ.get("DB_USER", "fabledcurator")
    password = os.environ["DB_PASSWORD"]
    host = os.environ.get("DB_HOST", "postgres")
    port = os.environ.get("DB_PORT", "5432")
    name = os.environ.get("DB_NAME", "fabledcurator_test")
    return f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{name}"


def _sync_database_url() -> str:
    user = os.environ.get("DB_USER", "fabledcurator")
    password = os.environ["DB_PASSWORD"]
    host = os.environ.get("DB_HOST", "postgres")
    port = os.environ.get("DB_PORT", "5432")
    name = os.environ.get("DB_NAME", "fabledcurator_test")
    return f"postgresql+psycopg://{user}:{password}@{host}:{port}/{name}"


@pytest_asyncio.fixture
async def engine():
    e = create_async_engine(_async_database_url(), future=True)
    yield e
    await e.dispose()


@pytest_asyncio.fixture
async def db(engine) -> AsyncSession:
    Session = async_sessionmaker(engine, expire_on_commit=False)
    async with Session() as session:
        await session.begin()
        try:
            yield session
        finally:
            await session.rollback()


@pytest.fixture
def sync_engine():
    e = create_engine(_sync_database_url(), future=True)
    yield e
    e.dispose()


@pytest.fixture
def db_sync(sync_engine):
    """Synchronous Session bound to a savepoint — used by Importer tests
    (the Importer is sync-only by design)."""
    SyncSession = sessionmaker(sync_engine, expire_on_commit=False)
    with SyncSession() as session:
        session.begin()
        try:
            yield session
        finally:
            session.rollback()


@pytest_asyncio.fixture
async def app():
    return create_app()


@pytest_asyncio.fixture
async def client(app):
    async with app.test_client() as c:
        yield c


# Migration-seeded baseline (singleton config like import_settings /
# ml_settings), captured lazily at the FIRST integration test's setup —
# when the CI integration job has just run `alembic upgrade head` so the DB
# is pristine. Cached module-wide. Never populated in the DB-less unit job
# because only integration-marked tests trigger the connection.
_SEED_SNAPSHOT: dict[str, list[dict]] | None = None


@pytest.fixture(scope="session")
def _truncate_engine():
    """Session-scoped sync engine reused by the per-test DB-reset teardown.

    Creating a fresh engine + Postgres connection for EVERY test's teardown was
    the integration suite's dominant cost — `--durations` showed the 15 slowest
    operations in both long shards were all ~1.5-2s teardowns, i.e. the
    connect+SCRAM handshake, not test logic. A single pooled connection reused
    across teardowns cuts that to the TRUNCATE itself. `pool_pre_ping` guards
    against Postgres reaping the idle connection mid-suite (a stale-connection
    failure would be a nasty flaky bounce; the ping is sub-millisecond).

    `create_engine` is lazy — it opens no connection until first `.begin()` —
    so the no-DB unit job instantiates this object but never connects.
    """
    eng = create_engine(_sync_database_url(), future=True, pool_pre_ping=True)
    yield eng
    eng.dispose()


@pytest.fixture(autouse=True)
def _reset_db_after_integration(request, _truncate_engine):
    """Integration tests exercise app/celery code that commits on its own
    connection, which the rollback fixtures above can't undo — so data
    would leak between tests. Capture the seeded baseline at the first
    integration test, then after each integration-marked test truncate
    every model table and restore that baseline. Touches the DB ONLY for
    integration-marked tests, so the no-DB fast unit job is untouched.

    Uses the session-scoped `_truncate_engine` (pooled, reused) rather than
    building a fresh engine per test — see that fixture's docstring.
    """
    global _SEED_SNAPSHOT
    is_integration = request.node.get_closest_marker("integration") is not None

    if is_integration and _SEED_SNAPSHOT is None:
        snap: dict[str, list[dict]] = {}
        with _truncate_engine.connect() as conn:
            for t in Base.metadata.sorted_tables:
                rows = [
                    dict(m)
                    for m in conn.execute(t.select()).mappings().all()
                ]
                if rows:
                    snap[t.name] = rows
        _SEED_SNAPSHOT = snap

    yield

    if not is_integration:
        return
    tables = ", ".join(t.name for t in Base.metadata.sorted_tables)
    if not tables:
        return
    with _truncate_engine.begin() as conn:
        conn.exec_driver_sql(
            f"TRUNCATE {tables} RESTART IDENTITY CASCADE"
        )
        for t in Base.metadata.sorted_tables:
            rows = (_SEED_SNAPSHOT or {}).get(t.name)
            if rows:
                conn.execute(t.insert(), rows)


@pytest_asyncio.fixture(autouse=True)
async def _dispose_app_engine(request):
    """The app's shared async engine (extensions.get_engine) is a process
    singleton, but pytest-asyncio gives each test a fresh event loop;
    asyncpg connections are loop-bound, so reuse across tests raises
    'attached to a different loop'. Dispose after each integration test so
    the next test rebuilds the engine on its own loop.
    """
    yield
    if request.node.get_closest_marker("integration") is None:
        return
    from backend.app import extensions
    await extensions.dispose_engine()