"""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, event, text from starlette.requests import HTTPConnection 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 # silent-ok / noqa BLE001: the question is literally "can anything reach it", so every # failure is the same answer — and this runs at import, before logging is configured. except Exception: # noqa: BLE001 return False finally: probe.dispose() _DB_REACHABLE = _db_reachable() # No background log writer in tests: every TestClient would otherwise start one on app startup and # write into tables the fixtures truncate between tests. The sink's own tests drive it explicitly. settings.log_db_sink = False @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." ) def _request_dep(db, unfinished: list[str]): """`get_db` for a test request: the test's own session, plus the check a shared session hides. A route helper flushes and its HANDLER commits (CLAUDE.md). Production's `get_db` closes the session after the request, which rolls back anything flushed but not committed — yet here every request shares the test's session, so a follow-up GET or an assertion reads the flushed write and the suite stays green over a handler that forgot to commit (deep review, debt burn). So a request that ends normally with a write it did not commit — flushed, or still only staged — is recorded, and the test fails on it at teardown. Only the ROOT transaction's end clears it: a savepoint released inside the handler commits nothing. The test's own staged rows are flushed before the request starts — they are the test's business, not the handler's. A request that raises is rolled back, as production's `close()` rolls it back: without that, what a failing handler flushed but never committed stayed readable to the rest of the test — the one state production cannot be in (E16: an issue created on the tracker before a failed upload was kept only by an explicit commit, and nothing saw it missing).""" def dep(conn: HTTPConnection): db.flush() state = {"dirty": False} def flushed(session, ctx): state["dirty"] = True def ended(session, transaction): if transaction.parent is None: state["dirty"] = False hooks = (("after_flush", flushed), ("after_transaction_end", ended)) for name, fn in hooks: event.listen(db, name, fn) try: yield db staged = db.new or db.deleted or any(db.is_modified(o) for o in db.dirty) if state["dirty"] or staged: unfinished.append(f"{conn.scope.get('method', 'WS')} {conn.url.path}") except Exception: db.rollback() raise finally: for name, fn in hooks: event.remove(db, name, fn) return dep @pytest.fixture def db(_schema): """A real session against the test Postgres, on a clean (empty) database. The code under test may commit freely. `db.info["request"]` is the `get_db` override for a request against it — the one every test client uses — and a handler that left a write uncommitted fails the test here.""" _truncate_all() session = SessionLocal() unfinished: list[str] = [] session.info["request"] = _request_dep(session, unfinished) session.info["unfinished"] = unfinished # the requests recorded so far (the check's own test) try: yield session finally: session.rollback() session.close() assert unfinished == [], "a handler left a write uncommitted" @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, db.info["request"]) 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, db.info["request"]) 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() @pytest.fixture(autouse=True) def _reset_log_erasures(): """`logs._erased` is module-level and remembers an erased account for an hour of monotonic time — longer than any suite run — while the `db` fixture hands serial ids back out from 1 via TRUNCATE … RESTART IDENTITY. So a test that purges user 1 would otherwise make a LATER test's rows for user 1 store with `user_id = None`, failing somewhere nowhere near the cause and only in some file orders. Same reasoning as the limiter sweep above, different global.""" from app import logs logs._erased.clear()