modbison/backend/tests/conftest.py

"""Shared pytest fixtures + hermetic-DB bootstrap.

The integration tests hit a real Postgres (Postgres-specific types: JSONB,
native ENUMs). To make ``cd backend && pytest`` work from a clean checkout
with no manual setup, ``pytest_configure`` below brings up an ephemeral
Postgres via docker compose and runs the migrations IF ``DATABASE_URL`` is
not already pointing at a reachable database. When it is (dev loop, or the
checkpoint runner which stands the DB up itself), that path is skipped and
the existing database is reused.

If Docker isn't available and no DB is reachable, the whole DB-backed suite
is skipped with a clear message rather than erroring out 60+ times.
"""

import asyncio
import os
import socket
import subprocess
import sys
import time
from pathlib import Path
from urllib.parse import urlsplit

import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.pool import NullPool

from app.config import settings
from app.database import get_db
from app.main import app

TEST_TABLES = ("avatar_loadouts", "assets", "worlds", "sessions", "refresh_tokens", "users")

BACKEND_DIR = Path(__file__).resolve().parents[1]
_COMPOSE_PROJECT = "agp-pytest"
_EPHEMERAL_PG_PORT = 25432

# Set by _ensure_database() when it starts a container it owns; torn down in
# pytest_unconfigure.
_started_container = False
_db_unavailable_reason: str | None = None


def _port_open(host: str, port: int, timeout: float = 1.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


def _db_reachable(url: str) -> bool:
    parts = urlsplit(url)
    return _port_open(parts.hostname or "localhost", parts.port or 5432)


def _run(cmd: list[str], **kw) -> subprocess.CompletedProcess:
    return subprocess.run(cmd, cwd=BACKEND_DIR, capture_output=True, text=True, **kw)


def pytest_configure(config):
    global _started_container, _db_unavailable_reason

    if _db_reachable(settings.DATABASE_URL):
        return  # dev loop / checkpoint runner already provided one

    if not _run(["docker", "--version"]).returncode == 0:
        _db_unavailable_reason = (
            "no reachable DATABASE_URL and docker is unavailable -- run the "
            "checkpoint suite (scripts/run_checkpoint_tests.py) or start Postgres "
            "manually"
        )
        return

    env = os.environ.copy()
    env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT
    env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT)
    env["POSTGRES_PORT"] = str(_EPHEMERAL_PG_PORT)
    up = _run(["docker", "compose", "up", "-d", "postgres"], env=env)
    if up.returncode != 0:
        _db_unavailable_reason = f"docker compose up postgres failed:\n{up.stderr[-800:]}"
        return
    _started_container = True

    for _ in range(30):
        r = _run(
            ["docker", "compose", "exec", "-T", "postgres",
             "pg_isready", "-U", "agp_user", "-d", "agp_db"],
            env=env,
        )
        if r.returncode == 0:
            break
        time.sleep(1)
    else:
        _db_unavailable_reason = "ephemeral Postgres never became ready"
        return

    url = f"postgresql+asyncpg://agp_user:agp_password@localhost:{_EPHEMERAL_PG_PORT}/agp_db"
    os.environ["DATABASE_URL"] = url
    settings.DATABASE_URL = url

    mig = _run([sys.executable, "-m", "alembic", "upgrade", "head"],
               env={**os.environ, "DATABASE_URL": url})
    if mig.returncode != 0:
        _db_unavailable_reason = f"alembic upgrade head failed:\n{mig.stderr[-800:]}"


def pytest_unconfigure(config):
    if not _started_container:
        return
    env = os.environ.copy()
    env["COMPOSE_PROJECT_NAME"] = _COMPOSE_PROJECT
    env["POSTGRES_HOST_PORT"] = str(_EPHEMERAL_PG_PORT)
    env["POSTGRES_PORT"] = str(_EPHEMERAL_PG_PORT)
    _run(["docker", "compose", "down", "-v"], env=env)


@pytest.fixture(scope="session", autouse=True)
def _reset_database():
    """Truncate all application tables once per test session."""
    if _db_unavailable_reason:
        pytest.skip(_db_unavailable_reason, allow_module_level=True)

    async def _truncate() -> None:
        engine = create_async_engine(settings.DATABASE_URL)
        async with engine.begin() as conn:
            await conn.execute(text(f"TRUNCATE TABLE {', '.join(TEST_TABLES)} CASCADE"))
        await engine.dispose()

    asyncio.run(_truncate())
    yield


@pytest.fixture(autouse=True)
def _override_db():
    """Override get_db so the app uses a fresh session per test.

    Deliberately does NOT reuse app.database's shared engine: FastAPI's
    (sync) TestClient runs each request through anyio's blocking-portal
    machinery, which can hand different requests different event loops.
    asyncpg connections are bound to the loop they were created on, so
    reusing one from the shared pool across requests raises "attached to a
    different loop" the moment a test makes more than one request. NullPool
    sidesteps it: every checkout is a brand-new connection.
    """
    test_engine = create_async_engine(settings.DATABASE_URL, poolclass=NullPool)
    test_session_factory = async_sessionmaker(test_engine, expire_on_commit=False)

    async def _get_db():
        async_session = test_session_factory()
        try:
            yield async_session
            await async_session.commit()
        except Exception:
            await async_session.rollback()
            raise
        finally:
            await async_session.close()

    app.dependency_overrides[get_db] = _get_db
    yield
    app.dependency_overrides.clear()