From c802b26406d31778edcfe89d85afcb70863bf906 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Thu, 4 Jun 2026 07:35:28 -0400 Subject: [PATCH] perf(ci): reuse a session-scoped engine for the per-test DB reset MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The autouse integration teardown created a fresh SQLAlchemy engine + Postgres connection for EVERY test, then disposed it — --durations showed the 15 slowest ops in both long shards were all ~1.5-2s teardowns (the connect+SCRAM handshake, not test logic). Hoist the truncate engine to a session-scoped, pool_pre_ping'd fixture so the pooled connection is reused across teardowns; the TRUNCATE+restore still runs per test, so isolation is unchanged. Lazy create_engine means the no-DB unit job instantiates but never connects. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/conftest.py | 69 ++++++++++++++++++++++++++++------------------- 1 file changed, 42 insertions(+), 27 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index da8bb9d..b8733e0 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -104,32 +104,51 @@ async def client(app): _SEED_SNAPSHOT: dict[str, list[dict]] | None = None +@pytest.fixture(scope="session") +def _truncate_engine(): + """Session-scoped sync engine reused by the per-test DB-reset teardown. + + Creating a fresh engine + Postgres connection for EVERY test's teardown was + the integration suite's dominant cost — `--durations` showed the 15 slowest + operations in both long shards were all ~1.5-2s teardowns, i.e. the + connect+SCRAM handshake, not test logic. A single pooled connection reused + across teardowns cuts that to the TRUNCATE itself. `pool_pre_ping` guards + against Postgres reaping the idle connection mid-suite (a stale-connection + failure would be a nasty flaky bounce; the ping is sub-millisecond). + + `create_engine` is lazy — it opens no connection until first `.begin()` — + so the no-DB unit job instantiates this object but never connects. + """ + eng = create_engine(_sync_database_url(), future=True, pool_pre_ping=True) + yield eng + eng.dispose() + + @pytest.fixture(autouse=True) -def _reset_db_after_integration(request): +def _reset_db_after_integration(request, _truncate_engine): """Integration tests exercise app/celery code that commits on its own connection, which the rollback fixtures above can't undo — so data would leak between tests. Capture the seeded baseline at the first integration test, then after each integration-marked test truncate - every model table and restore that baseline. Connects to the DB ONLY - for integration-marked tests, so the no-DB fast unit job is untouched. + every model table and restore that baseline. Touches the DB ONLY for + integration-marked tests, so the no-DB fast unit job is untouched. + + Uses the session-scoped `_truncate_engine` (pooled, reused) rather than + building a fresh engine per test — see that fixture's docstring. """ global _SEED_SNAPSHOT is_integration = request.node.get_closest_marker("integration") is not None if is_integration and _SEED_SNAPSHOT is None: - eng = create_engine(_sync_database_url(), future=True) snap: dict[str, list[dict]] = {} - try: - with eng.connect() as conn: - for t in Base.metadata.sorted_tables: - rows = [ - dict(m) - for m in conn.execute(t.select()).mappings().all() - ] - if rows: - snap[t.name] = rows - finally: - eng.dispose() + with _truncate_engine.connect() as conn: + for t in Base.metadata.sorted_tables: + rows = [ + dict(m) + for m in conn.execute(t.select()).mappings().all() + ] + if rows: + snap[t.name] = rows _SEED_SNAPSHOT = snap yield @@ -139,18 +158,14 @@ def _reset_db_after_integration(request): tables = ", ".join(t.name for t in Base.metadata.sorted_tables) if not tables: return - eng = create_engine(_sync_database_url(), future=True) - try: - with eng.begin() as conn: - conn.exec_driver_sql( - f"TRUNCATE {tables} RESTART IDENTITY CASCADE" - ) - for t in Base.metadata.sorted_tables: - rows = (_SEED_SNAPSHOT or {}).get(t.name) - if rows: - conn.execute(t.insert(), rows) - finally: - eng.dispose() + with _truncate_engine.begin() as conn: + conn.exec_driver_sql( + f"TRUNCATE {tables} RESTART IDENTITY CASCADE" + ) + for t in Base.metadata.sorted_tables: + rows = (_SEED_SNAPSHOT or {}).get(t.name) + if rows: + conn.execute(t.insert(), rows) @pytest_asyncio.fixture(autouse=True)