test(db): stand up the DB-backed pytest lane (R6 S3b)
The suite was pure-logic only (R2 S2b deferred the DB lane). Add an opt-in `db` fixture (tests/conftest.py) that hands a test a real session against an ephemeral Postgres; schema is built with `alembic upgrade head` (the real prod schema needs the unaccent extension + unaccent_simple TS config that create_all can't provide, and this also proves the migration chain applies). Each test starts from a TRUNCATEd clean DB so tests that COMMIT — the whole point, for R6 S2's transaction work — don't leak. backend/docker-compose.test.yml wires an ephemeral tmpfs pg + the test image. When no Postgres is reachable (a plain `docker run`), the DB tests SKIP and the pure-logic lane is unchanged: 172 passed, 2 skipped. With the DB: 174 passed. test_db_lane.py smoke-proves commit round-trip + TRUNCATE isolation. The gate wiring (siftlode.sh) lands in the ops repo.
This commit is contained in:
@@ -0,0 +1,35 @@
|
||||
# DB-backed pytest lane (R6 S3b): an ephemeral Postgres + the test image, wired together so the
|
||||
# `db` fixture (tests/conftest.py) has a real database. The whole thing is throwaway — the pg data
|
||||
# lives in tmpfs (RAM) and nothing is persisted. The pure-logic lane still runs WITHOUT this file
|
||||
# (plain `docker run … python -m pytest` — DB tests skip when no Postgres is reachable).
|
||||
#
|
||||
# docker compose -f backend/docker-compose.test.yml up --build --abort-on-container-exit \
|
||||
# --exit-code-from test
|
||||
# docker compose -f backend/docker-compose.test.yml down -v
|
||||
services:
|
||||
testdb:
|
||||
image: postgres:16-alpine
|
||||
environment:
|
||||
POSTGRES_USER: siftlode
|
||||
POSTGRES_PASSWORD: siftlode
|
||||
POSTGRES_DB: siftlode_test
|
||||
tmpfs:
|
||||
- /var/lib/postgresql/data # ephemeral + fast; the schema is rebuilt per run
|
||||
healthcheck:
|
||||
test: ["CMD-SHELL", "pg_isready -U siftlode -d siftlode_test"]
|
||||
interval: 1s
|
||||
timeout: 3s
|
||||
retries: 30
|
||||
test:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile.test
|
||||
depends_on:
|
||||
testdb:
|
||||
condition: service_healthy
|
||||
environment:
|
||||
DATABASE_URL: postgresql+psycopg://siftlode:siftlode@testdb:5432/siftlode_test
|
||||
volumes:
|
||||
- ./:/app
|
||||
working_dir: /app
|
||||
command: python -m pytest
|
||||
@@ -0,0 +1,65 @@
|
||||
"""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 pytest
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
from sqlalchemy import text
|
||||
|
||||
import app.models # noqa: F401 — register every model on Base.metadata
|
||||
from app.db import Base, SessionLocal, engine
|
||||
|
||||
|
||||
def _db_reachable() -> bool:
|
||||
try:
|
||||
with engine.connect() as conn:
|
||||
conn.execute(text("SELECT 1"))
|
||||
return True
|
||||
except Exception:
|
||||
return False
|
||||
|
||||
|
||||
_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"))
|
||||
|
||||
|
||||
@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()
|
||||
@@ -0,0 +1,21 @@
|
||||
"""Smoke test proving the DB-backed lane works (R6 S3b): a real commit round-trips, and each
|
||||
test starts from a clean database (TRUNCATE isolation), which is what lets the transaction-
|
||||
ownership tests (R6 S2) trust what they commit."""
|
||||
from sqlalchemy import func, select
|
||||
|
||||
from app.models import User
|
||||
|
||||
|
||||
def test_commit_roundtrips(db):
|
||||
db.add(User(email="lane@example.com", role="admin"))
|
||||
db.commit()
|
||||
|
||||
got = db.execute(select(User).where(User.email == "lane@example.com")).scalar_one()
|
||||
assert got.id is not None
|
||||
assert got.role == "admin"
|
||||
assert got.is_active is True # server_default applied on flush/commit
|
||||
|
||||
|
||||
def test_isolation_starts_clean(db):
|
||||
# If the previous test's committed row leaked, this count would be 1 — TRUNCATE must reset it.
|
||||
assert db.scalar(select(func.count()).select_from(User)) == 0
|
||||
Reference in New Issue
Block a user