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`
builds its connection in a fresh interpreter each time, unaffected by this
process's `lru_cache`d `get_settings()` — by the time a test runs, other test
modules have almost certainly already imported `src.main` and cached a
Settings instance pointed at production. `run_alembic` always passes
`-x target=test` (core-api T-3: alembic refuses to run with no target, or
any other one) and sets `TEST_DATABASE_URL` in the child's environment to
`url`, which is what actually redirects the migration — the subprocess's own
`alembic/env.py` reads that variable itself; nothing here overrides
`POSTGRES_*` any more.
"""
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 run_alembic(url: str, *args: str) -> subprocess.CompletedProcess:
    """Run `alembic -x target=test <args>` against `url`, never against the
    repo's own .env database."""
    env = os.environ.copy()
    env["TEST_DATABASE_URL"] = url
    result = subprocess.run(
        [str(ALEMBIC_BIN), "-x", "target=test", *args],
        cwd=str(REPO_ROOT),
        env=env,
        capture_output=True,
        text=True,
    )
    return result


class MaskedURL(str):
    """A database URL, fully usable as a plain string (passed to `.replace()`,
    `urlsplit()`, a subprocess environment, ...) but never through `repr()`
    (core-api T-12): pytest prints a failing test's fixture arguments using
    `repr()`, with no `-v`/`--showlocals` needed to trigger it — that is how
    a failing database test used to print this fixture's password straight
    into the terminal. `__str__` is deliberately left alone, so the real
    value is still what every caller actually connects with."""

    def __repr__(self) -> str:
        parts = urlsplit(self)
        netloc = parts.netloc
        if parts.password:
            netloc = netloc.replace(f":{parts.password}@", ":***@")
        return repr(parts._replace(netloc=netloc).geturl())


@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 MaskedURL(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 tables together so Postgres resolves
    the FK order (mailbox_entries -> messages, messages' self-referencing
    reply_to) across the one statement.

    `tenant_consumers` joined the list for migration 008 (core-api T-11):
    its new composite FK `(tenant_id, default_address)` ->
    `scribe_addresses(owner_tenant, name)` means Postgres now refuses to
    truncate `scribe_addresses` alone — "cannot truncate a table
    referenced in a foreign key constraint" — a restriction on the
    constraint's existence, not on whether any row's `default_address` is
    actually non-null (it never is here; migration 008's data step is a
    no-op in `core_api_test`). `TRUNCATE ... CASCADE` was rejected: it
    would silently drop every grant rather than only addresses, so the
    four grants this file's tests depend on — both seeded tenants x
    office-claude/office-gemini, the same pairing `_grant_test_office_gemini`
    establishes once per session — are reinserted immediately after."""
    from sqlalchemy import text

    await scribe_session.execute(
        text(
            "TRUNCATE scribe_mailbox_entries, scribe_messages, scribe_addresses, "
            "tenant_consumers"
        )
    )
    await scribe_session.execute(
        text(
            "INSERT INTO tenant_consumers (tenant_id, consumer) VALUES "
            "('jpmschweitzer', 'office-claude'), ('llm_tester', 'office-claude'), "
            "('jpmschweitzer', 'office-gemini'), ('llm_tester', 'office-gemini')"
        )
    )
    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()


@pytest.fixture
async def default_address_grant(tenant_registry: str, test_database_url: str):
    """
    Give `jpmschweitzer`/`office-claude` a default address (core-api T-11):
    registers a throwaway `scribe_addresses` row first — migration 008's
    composite FK means the grant's default must name a real address in the
    same tenant — then points the grant at it. Reversed on teardown (default
    back to null, the address row deleted) so the rest of the suite sees the
    null default that `core_api_test`'s own copy of migration 008 actually
    leaves this grant with (see test_tenants_migration.py's
    TestMigration008DataStep: claude-ai is never seeded here).
    """
    from sqlalchemy import text
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
    from sqlalchemy.pool import NullPool

    tenant_id = "jpmschweitzer"
    consumer = "office-claude"
    address = "test-default-address"

    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 scribe_addresses (owner_tenant, name, created_by) "
                "VALUES (:tenant, :address, 'test-harness')"
            ),
            {"tenant": tenant_id, "address": address},
        )
        await session.execute(
            text(
                "UPDATE tenant_consumers SET default_address = :address "
                "WHERE tenant_id = :tenant AND consumer = :consumer"
            ),
            {"tenant": tenant_id, "address": address, "consumer": consumer},
        )
        await session.commit()

    yield address

    async with session_factory() as session:
        await session.execute(
            text(
                "UPDATE tenant_consumers SET default_address = NULL "
                "WHERE tenant_id = :tenant AND consumer = :consumer"
            ),
            {"tenant": tenant_id, "consumer": consumer},
        )
        await session.execute(
            text("DELETE FROM scribe_addresses WHERE owner_tenant = :tenant AND name = :address"),
            {"tenant": tenant_id, "address": address},
        )
        await session.commit()
    await engine.dispose()