v2x-server/tests/conftest.py

"""Shared fixtures.

Tests run against a real MySQL database (`v2x_test`), never SQLite: this project
relies on MySQL types, collation and constraint behaviour, and a SQLite-backed
suite would happily pass while production broke.

Isolation strategy: each test runs inside an outer transaction that is rolled
back afterwards, so tests share one migrated schema without sharing data.
"""

from __future__ import annotations

import os
from collections.abc import AsyncIterator, Callable, Iterator
from contextlib import AbstractAsyncContextManager
from typing import Any

import pytest
from alembic import command
from alembic.config import Config
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession, async_sessionmaker

from v2x_server.auth.deps import get_current_principal
from v2x_server.auth.principal import Principal
from v2x_server.core.config import Settings
from v2x_server.db.session import create_engine, get_session
from v2x_server.main import create_app

PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))


@pytest.fixture(scope="session")
def settings() -> Settings:
    base = Settings()
    return base.model_copy(
        update={
            "app_env": "test",
            "database_url": base.test_database_url,
            "cors_origins": [],
        }
    )


@pytest.fixture(scope="session")
def _migrated_database(settings: Settings) -> Iterator[None]:
    """Bring `v2x_test` to head once per session."""
    config = Config(os.path.join(PROJECT_ROOT, "alembic.ini"))
    config.set_main_option("script_location", os.path.join(PROJECT_ROOT, "migrations"))
    os.environ["ALEMBIC_DATABASE_URL"] = settings.database_url

    command.upgrade(config, "head")
    yield
    os.environ.pop("ALEMBIC_DATABASE_URL", None)


@pytest.fixture
async def db_connection(
    settings: Settings, _migrated_database: None
) -> AsyncIterator[AsyncConnection]:
    """A connection with an open outer transaction, rolled back after the test."""
    engine = create_engine(settings)
    async with engine.connect() as connection:
        transaction = await connection.begin()
        try:
            yield connection
        finally:
            await transaction.rollback()
    await engine.dispose()


@pytest.fixture
async def db_session(db_connection: AsyncConnection) -> AsyncIterator[AsyncSession]:
    """Session bound to the test's connection, joined to its outer transaction.

    `join_transaction_mode="create_savepoint"` lets application code call
    commit() normally -- it releases a savepoint rather than committing the
    outer transaction, so the rollback above still wipes everything.
    """
    factory = async_sessionmaker(
        bind=db_connection,
        expire_on_commit=False,
        join_transaction_mode="create_savepoint",
    )
    async with factory() as session:
        yield session


class LifespanRunner:
    """Minimal async context manager that drives an app's lifespan events."""

    def __init__(self, app: FastAPI) -> None:
        self._app = app
        self._context: AbstractAsyncContextManager[None] | None = None

    async def __aenter__(self) -> None:
        self._context = self._app.router.lifespan_context(self._app)
        await self._context.__aenter__()

    async def __aexit__(self, *exc_info: Any) -> None:
        assert self._context is not None
        await self._context.__aexit__(*exc_info)


def make_principal(*roles: str, username: str = "test-user") -> Principal:
    return Principal(
        subject="00000000-0000-0000-0000-000000000001",
        username=username,
        email=f"{username}@example.com",
        realm_roles=frozenset(roles),
    )


@pytest.fixture
async def app(settings: Settings, db_session: AsyncSession) -> AsyncIterator[FastAPI]:
    """App wired to the test session; auth left unauthenticated by default."""
    application = create_app(settings)

    async def _override_session() -> AsyncIterator[AsyncSession]:
        yield db_session

    application.dependency_overrides[get_session] = _override_session
    yield application
    application.dependency_overrides.clear()


@pytest.fixture
def as_user(app: FastAPI) -> Callable[..., Principal]:
    """Authenticate subsequent requests as a principal holding `roles`."""

    def _apply(*roles: str, username: str = "test-user") -> Principal:
        principal = make_principal(*roles, username=username)
        app.dependency_overrides[get_current_principal] = lambda: principal
        return principal

    return _apply


@pytest.fixture
async def client(app: FastAPI) -> AsyncIterator[AsyncClient]:
    """In-process HTTP client. No live server, but the full middleware stack."""
    transport = ASGITransport(app=app)
    # LifespanRunner is what builds the engine and OIDC provider, so app.state
    # matches production rather than being half-initialised.
    async with (
        LifespanRunner(app),
        AsyncClient(transport=transport, base_url="http://test") as http_client,
    ):
        yield http_client