modbison/backend/tests/conftest.py

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

The integration tests hit a real Postgres (Postgres-specific types: JSONB,
native ENUMs) and, for presence tests, a real Redis (TTL semantics aren't
worth faking). To make ``cd backend && pytest`` work from a clean checkout
with no manual setup, ``pytest_configure`` below brings up ephemeral
Postgres/Redis containers via docker compose and runs the migrations IF
``DATABASE_URL``/``REDIS_URL`` aren't already pointing at something
reachable. When they are (dev loop, or the checkpoint runner which stands
these up itself), that path is skipped and the existing services are
reused.

If Docker isn't available and nothing is reachable, the whole
DB/Redis-backed suite is skipped with a clear message rather than erroring
out dozens of 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",
    # world_chunk_revisions has no FK to worlds/world_chunks (world.md §4.7 /
    # world_persistence_impl.md §4.3, deliberate -- deleting a chunk or world
    # must not lose the audit trail), so it won't be cascade-truncated via
    # `worlds` the way world_rooms/world_chunks would be. List all three
    # explicitly rather than rely on cascade for some of them.
    "world_chunk_revisions", "world_chunks", "world_rooms",
    "worlds", "sessions", "refresh_tokens", "users",
)

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

# Set by pytest_configure() when it starts container(s) 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 _redis_reachable(url: str) -> bool:
    parts = urlsplit(url)
    return _port_open(parts.hostname or "localhost", parts.port or 6379)


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

    need_db = not _db_reachable(settings.DATABASE_URL)
    need_redis = not _redis_reachable(settings.REDIS_URL)

    if not need_db and not need_redis:
        return  # dev loop / checkpoint runner already provided both

    if not _run(["docker", "--version"]).returncode == 0:
        _db_unavailable_reason = (
            "no reachable DATABASE_URL/REDIS_URL and docker is unavailable -- run "
            "the checkpoint suite (scripts/run_checkpoint_tests.py) or start "
            "Postgres/Redis 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)
    env["REDIS_HOST_PORT"] = str(_EPHEMERAL_REDIS_PORT)

    services = [s for s, need in (("postgres", need_db), ("redis", need_redis)) if need]
    up = _run(["docker", "compose", "up", "-d", *services], env=env)
    if up.returncode != 0:
        _db_unavailable_reason = f"docker compose up {' '.join(services)} failed:\n{up.stderr[-800:]}"
        return
    _started_container = True

    if need_db:
        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:]}"
            return

    if need_redis:
        for _ in range(30):
            r = _run(
                ["docker", "compose", "exec", "-T", "redis", "redis-cli", "ping"],
                env=env,
            )
            if r.returncode == 0 and "PONG" in r.stdout:
                break
            time.sleep(1)
        else:
            _db_unavailable_reason = "ephemeral Redis never became ready"
            return

        redis_url = f"redis://localhost:{_EPHEMERAL_REDIS_PORT}/0"
        os.environ["REDIS_URL"] = redis_url
        settings.REDIS_URL = redis_url


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)
    env["REDIS_HOST_PORT"] = str(_EPHEMERAL_REDIS_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 _reset_rate_limiter():
    """Every TestClient request looks like the same client IP, so the
    per-IP auth limiter would trip mid-suite. Clear it between tests; the
    dedicated rate-limit test disables this to exercise the real behaviour.
    """
    from app import rate_limit
    rate_limit.reset_all()
    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()