core-api/tests/conftest.py

"""
Shared test fixtures.

Database-backed tests need `TEST_DATABASE_URL` (core-api's gitignored .env).
They **skip** when it is unset and **refuse** (hard failure, not a skip)
when it points at `core_api` — that is production, and the dev `.env`'s
other `POSTGRES_*` values point there too, so this is the one guard that
stands between a test run and the production database.

Running the actual migration against the test database is done through the
`alembic` CLI in a subprocess rather than in-process, because
`alembic/env.py` always builds its connection URL from `get_settings()`
(composed from `POSTGRES_*`), with no override hook of its own — and
`get_settings()` is `lru_cache`d, so by the time a test runs, other test
modules have almost certainly already imported `src.main` and cached a
Settings instance pointed at production. A subprocess gets a clean
environment and a clean cache; overriding `POSTGRES_HOST`/`USER`/
`PASSWORD`/`DATABASE` in that child's environment, derived from
`TEST_DATABASE_URL`, is what actually redirects the migration.
"""
import os
import subprocess
from pathlib import Path
from urllib.parse import urlsplit

import pytest
from dotenv import load_dotenv

REPO_ROOT = Path(__file__).resolve().parent.parent
ALEMBIC_BIN = REPO_ROOT / ".venv" / "bin" / "alembic"

load_dotenv(REPO_ROOT / ".env")


def _components(url: str) -> dict:
    """POSTGRES_* overrides for a subprocess, derived from a database URL."""
    parts = urlsplit(url)
    host = parts.hostname or ""
    if parts.port:
        host = f"{host}:{parts.port}"
    return {
        "POSTGRES_HOST": host,
        "POSTGRES_USER": parts.username or "",
        "POSTGRES_PASSWORD": parts.password or "",
        "POSTGRES_DATABASE": parts.path.lstrip("/"),
    }


def run_alembic(url: str, *args: str) -> subprocess.CompletedProcess:
    """Run `alembic <args>` against `url`, never against the repo's own .env database."""
    env = os.environ.copy()
    env.update(_components(url))
    result = subprocess.run(
        [str(ALEMBIC_BIN), *args],
        cwd=str(REPO_ROOT),
        env=env,
        capture_output=True,
        text=True,
    )
    return result


@pytest.fixture(scope="session")
def test_database_url() -> str:
    """The core_api_test database URL, or skip/refuse per the rules above."""
    url = os.environ.get("TEST_DATABASE_URL")
    if not url:
        pytest.skip("TEST_DATABASE_URL is not set — database tests skipped")

    db_name = urlsplit(url).path.lstrip("/")
    if db_name == "core_api":
        pytest.fail(
            "TEST_DATABASE_URL resolves to 'core_api' (production) — refusing to use it for tests"
        )

    return url


@pytest.fixture(scope="session")
def book_schema(test_database_url: str) -> str:
    """Ensure migration 005 (and everything before it) is applied to core_api_test."""
    result = run_alembic(test_database_url, "upgrade", "head")
    if result.returncode != 0:
        pytest.fail(f"alembic upgrade head failed:\n{result.stdout}\n{result.stderr}")
    return test_database_url


@pytest.fixture
async def book_session(book_schema: str):
    """An AsyncSession bound to core_api_test, rolled back after each test."""
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
    from sqlalchemy.pool import NullPool

    asyncpg_url = book_schema.replace("postgresql://", "postgresql+asyncpg://", 1)
    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
    session_factory = async_sessionmaker(bind=engine, expire_on_commit=False)

    async with session_factory() as session:
        yield session
        await session.rollback()

    await engine.dispose()


@pytest.fixture
async def clean_book_tables(book_session):
    """
    Book service methods commit internally (same pattern as the Dashboard
    service), so `book_session`'s post-test rollback never runs against
    anything — the data is already durable. TRUNCATE before each test
    instead, which the append-only trigger does not block: it is a
    row-level BEFORE UPDATE OR DELETE trigger, and TRUNCATE fires neither.
    """
    from sqlalchemy import text

    await book_session.execute(
        text("TRUNCATE book_events, book_attachments, book_record_links, book_tags, book_records")
    )
    await book_session.commit()
    yield book_session


@pytest.fixture(scope="session")
def scribe_schema(test_database_url: str) -> str:
    """Ensure migration 006 (and everything before it) is applied to core_api_test."""
    result = run_alembic(test_database_url, "upgrade", "head")
    if result.returncode != 0:
        pytest.fail(f"alembic upgrade head failed:\n{result.stdout}\n{result.stderr}")
    return test_database_url


@pytest.fixture
async def scribe_session(scribe_schema: str):
    """An AsyncSession bound to core_api_test — same per-test engine pattern
    as book_session (see its docstring for why: TestClient drives the app
    on a different event loop than pytest-asyncio's fixture setup)."""
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
    from sqlalchemy.pool import NullPool

    asyncpg_url = scribe_schema.replace("postgresql://", "postgresql+asyncpg://", 1)
    engine = create_async_engine(asyncpg_url, poolclass=NullPool)
    session_factory = async_sessionmaker(bind=engine, expire_on_commit=False)

    async with session_factory() as session:
        yield session
        await session.rollback()

    await engine.dispose()


@pytest.fixture
async def clean_scribe_tables(scribe_session):
    """Scribe service methods commit internally (same reason as Book's
    clean_book_tables). TRUNCATE all three tables together so Postgres
    resolves the FK order (mailbox_entries -> messages, messages'
    self-referencing reply_to) across the one statement."""
    from sqlalchemy import text

    await scribe_session.execute(
        text("TRUNCATE scribe_mailbox_entries, scribe_messages, scribe_addresses")
    )
    await scribe_session.commit()
    yield scribe_session