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.testclient import TestClient
from sqlalchemy import create_engine, text

import app.models  # noqa: F401 — register every model on Base.metadata
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

# 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 client(db):
    """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)."""
    app.dependency_overrides[get_db] = lambda: db
    try:
        yield TestClient(app)
    finally:
        app.dependency_overrides.clear()


@pytest.fixture
def as_user(db):
    """Factory -> a TestClient already authenticated as a given User (via a `current_user` override),
    for tests that need an authed request but aren't exercising the login flow itself."""
    def _as(user: User) -> TestClient:
        app.dependency_overrides[get_db] = lambda: db
        app.dependency_overrides[current_user] = lambda: user
        return TestClient(app)

    try:
        yield _as
    finally:
        app.dependency_overrides.clear()