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 asyncio
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


def _grant_test_office_gemini(url: str) -> None:
    """
    Grant a second, test-environment-only consumer, `office-gemini`, in
    both seeded tenants: test_book.py and test_scribe.py have used
    `office-gemini` as a stand-in second Office consumer since before
    core-api T-8's grant check existed, to prove behaviour that depends on
    two *distinct* consumers (address ownership across consumers,
    `book_events.client` recording the actual caller) rather than on any
    one consumer being granted. Migration 007 itself seeds only
    `office-claude` — the only consumer actually configured in Authentik
    today (T-8's dossier) — so this grant is inserted directly here,
    test-only, rather than through the migration; production's seed stays
    exactly what the dossier specifies. A consumer this never grants
    (e.g. `office-ungranted` in a test) is what proves the
    `consumer_not_granted` refusal — this must never widen to cover one of
    those.

    A plain function, not a fixture: called from each of `book_schema`,
    `scribe_schema` and `tenant_registry` right after their own `run_alembic
    upgrade head`, so the grant is restored whenever any one of them is the
    first to re-establish the schema in a session — the same
    self-healing-on-first-use property those fixtures already rely on to
    survive a migration test downgrading the schema mid-suite. A session-
    scoped fixture shared between them would cache the *string* across that
    downgrade while the real schema moved on underneath it — which is what
    a first version of this did, and `tests/test_scribe.py` failing with
    "relation tenants does not exist" is what caught it.
    """
    from sqlalchemy import text
    from sqlalchemy.ext.asyncio import create_async_engine
    from sqlalchemy.pool import NullPool

    asyncpg_url = url.replace("postgresql://", "postgresql+asyncpg://", 1)

    async def _grant() -> None:
        engine = create_async_engine(asyncpg_url, poolclass=NullPool)
        try:
            async with engine.begin() as conn:
                await conn.execute(
                    text(
                        "INSERT INTO tenant_consumers (tenant_id, consumer) VALUES "
                        "('jpmschweitzer', 'office-gemini'), ('llm_tester', 'office-gemini') "
                        "ON CONFLICT DO NOTHING"
                    )
                )
        finally:
            await engine.dispose()

    asyncio.run(_grant())


@pytest.fixture(scope="session")
def tenant_registry(test_database_url: str) -> str:
    """Ensure migration 007 (and everything before it) is applied to
    core_api_test, with its seed tenants/grants plus the test-only
    office-gemini grant (see `_grant_test_office_gemini`). Used directly by
    tests/test_tenants.py; book_schema and scribe_schema do the same thing
    independently rather than depending on this, so each keeps its own
    self-healing-on-first-use property (see that function's docstring)."""
    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}")
    _grant_test_office_gemini(test_database_url)
    return test_database_url


@pytest.fixture(scope="session")
def book_schema(test_database_url: str) -> str:
    """Ensure migration 005 (and everything before it, now including the
    tenant registry) 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}")
    _grant_test_office_gemini(test_database_url)
    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, book_tenant_counters"
        )
    )
    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, now including the
    tenant registry) 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}")
    _grant_test_office_gemini(test_database_url)
    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


@pytest.fixture
async def suspended_tenant(test_database_url: str):
    """
    A third tenant, inserted with `status='suspended'`, for proving the
    `tenant_suspended` refusal (core-api T-8) without touching either
    seeded tenant that the rest of the suite depends on being active.
    Nothing else ever references this id, so inserting and deleting it
    per test is safe — no FK points at it.

    Depends on `test_database_url` and runs its own `alembic upgrade head`
    rather than on `tenant_registry`/`book_schema`/`scribe_schema`: this
    fixture is reused from test_book.py, test_scribe.py and test_tenants.py,
    so whichever of those files runs first would permanently cache
    *any* session-scoped schema fixture it depended on — and a migration
    test downgrading the schema afterwards (test_book_migration.py,
    test_scribe_migration.py) would leave that cache stale for the other
    two files. Self-contained and function-scoped avoids it outright, at
    the cost of a redundant `alembic upgrade head` call (cheap relative to
    the many such calls the migration test files already make).
    """
    from sqlalchemy import text
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
    from sqlalchemy.pool import NullPool

    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}")

    tenant_id = "suspended-tenant"
    asyncpg_url = test_database_url.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:
        await session.execute(
            text(
                "INSERT INTO tenants (id, display_name, kind, status) "
                "VALUES (:id, 'Suspended Test Tenant', 'test', 'suspended') "
                "ON CONFLICT (id) DO UPDATE SET status = 'suspended'"
            ),
            {"id": tenant_id},
        )
        await session.commit()

    yield tenant_id

    async with session_factory() as session:
        await session.execute(text("DELETE FROM tenants WHERE id = :id"), {"id": tenant_id})
        await session.commit()
    await engine.dispose()