test(fixtures): make the TestClient fixtures isolated and composable (S2 review)

Address the /code-review findings on the S2 test-net:
- as_user no longer sets a global `lambda: user` override that two users in one
  test would collide on; the shared current_user override now resolves the caller
  from an x-test-user header, and each returned client tags its own requests, so
  as_user(a)/as_user(b) are independent (#1).
- override_dep records the prior value per dependency key and restores it on
  teardown instead of a blanket dependency_overrides.clear(), so fixtures compose
  without clobbering each other (#5).
- a _configured fixture bakes the post-setup state into client/as_user, and the
  per-test inline monkeypatch / autouse duplicates are removed (#4).
- the feed keyset walk asserts all seeded rows surfaced before comparing, so a
  seeding regression can't make the sorts pass vacuously (#3).
This commit is contained in:
2026-08-18 01:39:45 +02:00
parent 296af2a498
commit e764ca4568
3 changed files with 63 additions and 30 deletions
+56 -17
View File
@@ -18,10 +18,12 @@ import shutil
import pytest import pytest
from alembic import command from alembic import command
from alembic.config import Config from alembic.config import Config
from fastapi import HTTPException, Request
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sqlalchemy import create_engine, text from sqlalchemy import create_engine, text
import app.models # noqa: F401 — register every model on Base.metadata import app.models # noqa: F401 — register every model on Base.metadata
import app.state as app_state
from app import security from app import security
from app.auth import current_user from app.auth import current_user
from app.config import settings from app.config import settings
@@ -29,6 +31,9 @@ from app.db import Base, SessionLocal, engine, get_db
from app.main import app from app.main import app
from app.models import User 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 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. # — 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" _REQUIRE_FULL = os.getenv("SIFTLODE_TEST_REQUIRE_DB") == "1"
@@ -124,28 +129,62 @@ def make_user(db):
@pytest.fixture @pytest.fixture
def client(db): 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 """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. `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 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).""" the next on this same instance. No lifespan (the scheduler / HLS sweep stay off)."""
app.dependency_overrides[get_db] = lambda: db override_dep(get_db, lambda: db)
try: return TestClient(app)
yield TestClient(app)
finally:
app.dependency_overrides.clear()
@pytest.fixture @pytest.fixture
def as_user(db): def as_user(db, override_dep, _configured):
"""Factory -> a TestClient already authenticated as a given User (via a `current_user` override), """Factory -> a TestClient authenticated as a given User (bypassing login), for tests that need an
for tests that need an authed request but aren't exercising the login flow itself.""" authed request but aren't exercising the login flow itself. Multiple users in ONE test are isolated:
def _as(user: User) -> TestClient: the shared `current_user` override resolves the caller from an `x-test-user` header, and each
app.dependency_overrides[get_db] = lambda: db returned client tags its own requests with it — so `as_user(a)` and `as_user(b)` don't collide on
app.dependency_overrides[current_user] = lambda: user the process-global override the way a plain `lambda: user` would."""
return TestClient(app) registry: dict[int, User] = {}
try: def _resolver(request: Request) -> User:
yield _as raw = request.headers.get("x-test-user")
finally: if raw is None or (uid := int(raw)) not in registry:
app.dependency_overrides.clear() 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
+1 -7
View File
@@ -11,20 +11,14 @@ Note: `/api/me` is the bootstrap probe — it returns 200 `{authenticated: False
""" """
from datetime import timedelta from datetime import timedelta
import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
import app.state as state
from app.auth import ACTIVE_ACCOUNT_HEADER from app.auth import ACTIVE_ACCOUNT_HEADER
from app.main import app from app.main import app
_PW = "correct horse battery" _PW = "correct horse battery"
# The `client`/`as_user` fixtures bake in the post-setup `_configured` state, so no per-test patch here.
@pytest.fixture(autouse=True)
def _configured(monkeypatch):
# The setup gate 503s /auth and /api until the wizard has run; every test here is post-setup.
monkeypatch.setattr(state, "_configured_cache", True)
def _login(client, email, password): def _login(client, email, password):
+6 -6
View File
@@ -15,7 +15,6 @@ from datetime import datetime, timezone
import pytest import pytest
import app.state as state
from app.models import Channel, Subscription, User, Video from app.models import Channel, Subscription, User, Video
# Every sort feedquery.sort_keys defines except "relevance" (which needs a search query to rank). # Every sort feedquery.sort_keys defines except "relevance" (which needs a search query to rank).
@@ -74,8 +73,7 @@ def _page(client, sort, cursor):
return resp.json() return resp.json()
def test_feed_surfaces_the_whole_seed(db, as_user, monkeypatch): def test_feed_surfaces_the_whole_seed(db, as_user):
monkeypatch.setattr(state, "_configured_cache", True) # the request gate 503s an unconfigured app
user, ids = _seed(db) user, ids = _seed(db)
body = as_user(user).get("/api/feed?limit=50").json() body = as_user(user).get("/api/feed?limit=50").json()
assert {it["id"] for it in body["items"]} == set(ids) assert {it["id"] for it in body["items"]} == set(ids)
@@ -83,13 +81,15 @@ def test_feed_surfaces_the_whole_seed(db, as_user, monkeypatch):
@pytest.mark.parametrize("sort", _SORTS) @pytest.mark.parametrize("sort", _SORTS)
def test_cursor_walk_matches_the_unpaged_order(db, as_user, monkeypatch, sort): def test_cursor_walk_matches_the_unpaged_order(db, as_user, sort):
monkeypatch.setattr(state, "_configured_cache", True) user, ids = _seed(db)
user, _ = _seed(db)
client = as_user(user) client = as_user(user)
# Ground truth: the full order in one query (no cursor). # Ground truth: the full order in one query (no cursor).
full = [it["id"] for it in client.get(f"/api/feed?sort={sort}&limit=50").json()["items"]] full = [it["id"] for it in client.get(f"/api/feed?sort={sort}&limit=50").json()["items"]]
# Guard against a vacuous pass: if seeding ever stops surfacing the videos, `walked == full` would
# be [] == [] for every sort and never exercise the cursor. All 7 rows must be present.
assert len(full) == len(ids), f"{sort}: seed did not surface all rows — the walk would be vacuous"
# Walk it two at a time via the cursor and reassemble. # Walk it two at a time via the cursor and reassemble.
walked: list[str] = [] walked: list[str] = []