siftlode/backend/tests/conftest.py

"""Pytest fixtures for the DB-backed integration lane (R6 S3b).

The bulk of the suite is pure-logic and needs no database. This adds an OPT-IN `db` fixture
that hands a test a real session against a throwaway Postgres (the app's own engine, bound to
the test `DATABASE_URL`). When no database is reachable — the default `siftlode check` still
runs the pure-logic tests with no Postgres — any test that asks for `db` is SKIPPED, so the
pure-logic lane is unaffected.

Schema comes from `alembic upgrade head` — not `create_all` — because the real schema depends on
things only migrations set up (the `unaccent` extension + the `public.unaccent_simple` text-search
config the videos.search_vector generated column needs). Building it via migrations also validates
that the migration chain applies cleanly. Each test then gets a clean database via TRUNCATE so
tests that COMMIT (the whole point of this lane — transaction-ownership behaviour) don't leak.
"""

import os
import shutil

import pytest
from alembic import command
from alembic.config import Config
from fastapi import HTTPException, Request
from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text

import app.models  # noqa: F401 — register every model on Base.metadata
import app.state as app_state
from app import security
from app.auth import current_user
from app.config import settings
from app.db import Base, SessionLocal, engine, get_db
from app.main import app
from app.models import User

# Sentinel for "this dependency had no override before the fixture set one" (see override_dep).
_MISSING = object()

# The compose test lane sets this so a would-be silent skip (no DB, or no ffmpeg) becomes a hard error
# — the gate must not print green while the DB-backed / ffmpeg tests were quietly not run.
_REQUIRE_FULL = os.getenv("SIFTLODE_TEST_REQUIRE_DB") == "1"


def _db_reachable() -> bool:
    # A throwaway engine with a bounded connect_timeout so a slow/firewalled DATABASE_URL can't
    # hang the whole suite at import (the app engine has no connect timeout). The compose lane
    # gates on `service_healthy`, but this keeps a manual run honest too.
    probe = create_engine(settings.database_url, connect_args={"connect_timeout": 3})
    try:
        with probe.connect() as conn:
            conn.execute(text("SELECT 1"))
        return True
    except Exception:
        return False
    finally:
        probe.dispose()


_DB_REACHABLE = _db_reachable()


@pytest.fixture(scope="session")
def _schema():
    if not _DB_REACHABLE:
        pytest.skip("no test database (set DATABASE_URL to a reachable Postgres)")
    # Build the real prod schema via migrations (alembic env.py reads settings.database_url, i.e.
    # the test DATABASE_URL). Idempotent — a re-run against an already-migrated DB is a no-op.
    command.upgrade(Config("alembic.ini"), "head")
    yield


def _truncate_all():
    # RESTART IDENTITY so serial PKs are deterministic per test; CASCADE handles FKs. Truncating
    # BEFORE each test (not after) means every test — including the first, which would otherwise see
    # migration-seeded rows — starts from the same empty slate and seeds only what it needs.
    tables = ", ".join(t.name for t in Base.metadata.sorted_tables)
    with engine.begin() as conn:
        conn.execute(text(f"TRUNCATE {tables} RESTART IDENTITY CASCADE"))


def pytest_configure(config):
    """Fail the whole run — loudly, once — when SIFTLODE_TEST_REQUIRE_DB=1 but the environment can't
    actually run the full suite. Without this, an unreachable DATABASE_URL or a missing ffmpeg turns
    ~60 DB tests + the poster tests into silent skips and the gate still exits green."""
    if not _REQUIRE_FULL:
        return
    if not _DB_REACHABLE:
        raise pytest.UsageError(
            "SIFTLODE_TEST_REQUIRE_DB=1 but no Postgres is reachable at the test DATABASE_URL — the "
            "DB-backed lane would silently skip. Fix the DB wiring or unset the flag."
        )
    if shutil.which("ffmpeg") is None:
        raise pytest.UsageError(
            "SIFTLODE_TEST_REQUIRE_DB=1 but ffmpeg is missing — the thumbnail/poster tests would skip."
        )


@pytest.fixture
def db(_schema):
    """A real session against the test Postgres, on a clean (empty) database. The code under test
    may commit freely."""
    _truncate_all()
    session = SessionLocal()
    try:
        yield session
    finally:
        session.rollback()
        session.close()


@pytest.fixture
def make_user(db):
    """Factory: create + commit a User in the test DB. Defaults produce a login-ready account (email
    verified, active, with a known password); override any of it via kwargs (e.g. `role="admin"`,
    `is_suspended=True`, `password=None` for a Google-only account). Returns the User."""
    seq = {"n": 0}

    def _make(email: str | None = None, *, password: str | None = "correct horse battery", **kw):
        seq["n"] += 1
        kw.setdefault("email_verified", True)
        user = User(
            email=email or f"user{seq['n']}@example.com",
            password_hash=security.hash_password(password) if password else None,
            **kw,
        )
        db.add(user)
        db.commit()
        return user

    return _make


@pytest.fixture
def override_dep():
    """Set app.dependency_overrides for the duration of a test and restore the PRIOR value per key on
    teardown — instead of a blanket `app.dependency_overrides.clear()`, which wipes every override and
    stops fixtures composing (one fixture's teardown clobbering another's)."""
    saved: dict = {}

    def _set(dep, override):
        if dep not in saved:  # remember the pre-fixture value the FIRST time we touch this key
            saved[dep] = app.dependency_overrides.get(dep, _MISSING)
        app.dependency_overrides[dep] = override

    yield _set
    for dep, prev in saved.items():
        if prev is _MISSING:
            app.dependency_overrides.pop(dep, None)
        else:
            app.dependency_overrides[dep] = prev


@pytest.fixture
def _configured(monkeypatch):
    """The setup gate 503s /auth and /api until the wizard has run; the request-issuing fixtures below
    are all post-setup. Baked into client/as_user so no individual test has to remember it."""
    monkeypatch.setattr(app_state, "_configured_cache", True)


@pytest.fixture
def client(db, override_dep, _configured):
    """A TestClient wired to the test DB that goes through the REAL auth/session stack — no
    `current_user` override — so login, the wallet gate and epoch revoke are exercised end to end.
    Starlette's SessionMiddleware round-trips the signed cookie, so a login in one call authenticates
    the next on this same instance. No lifespan (the scheduler / HLS sweep stay off)."""
    override_dep(get_db, lambda: db)
    return TestClient(app)


@pytest.fixture
def as_user(db, override_dep, _configured):
    """Factory -> a TestClient authenticated as a given User (bypassing login), for tests that need an
    authed request but aren't exercising the login flow itself. Multiple users in ONE test are isolated:
    the shared `current_user` override resolves the caller from an `x-test-user` header, and each
    returned client tags its own requests with it — so `as_user(a)` and `as_user(b)` don't collide on
    the process-global override the way a plain `lambda: user` would."""
    registry: dict[int, User] = {}

    def _resolver(request: Request) -> User:
        raw = request.headers.get("x-test-user")
        if raw is None or (uid := int(raw)) not in registry:
            raise HTTPException(status_code=401, detail="Not authenticated")
        return registry[uid]

    override_dep(get_db, lambda: db)
    override_dep(current_user, _resolver)

    def _as(user: User) -> TestClient:
        registry[user.id] = user
        return TestClient(app, headers={"x-test-user": str(user.id)})

    return _as


@pytest.fixture(autouse=True)
def _reset_rate_limiters():
    """Every RateLimiter is a module-level singleton keyed by client IP, and TestClient always
    presents the same host — so without this the whole suite's login/register attempts pool into one
    300s window and, past the cap, some unrelated test starts failing with a 429 depending on order.
    Cleared before each test; a test that wants to exercise throttling can still exhaust one itself.

    Sweeps every loaded `app.*` module rather than a hand-kept list, so a limiter added in a new
    module is covered the day it lands instead of silently pooling until something fails."""
    import sys

    from app.ratelimit import RateLimiter

    for name, mod in list(sys.modules.items()):
        if name != "app" and not name.startswith("app."):
            continue
        for obj in vars(mod).values():
            if isinstance(obj, RateLimiter):
                obj.reset()