From 7a896605e031c4f51c5059c8cffd9adebb3bb5c6 Mon Sep 17 00:00:00 2001
From: Bryan Van Deusen <bryan.vandeusen@fabledsword.com>
Date: Fri, 15 May 2026 22:35:37 -0400
Subject: [PATCH] fix(integration): snapshot/restore seeded singletons, dispose
 app engine per test

---
 tests/conftest.py | 46 ++++++++++++++++++++++++++++++++++++++++++----
 1 file changed, 42 insertions(+), 4 deletions(-)

diff --git a/tests/conftest.py b/tests/conftest.py
index 619f4b0..e79b80c 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -77,13 +77,32 @@ def db_sync(sync_engine):
             session.rollback()
 
 
+@pytest.fixture(scope="session")
+def _seed_snapshot():
+    """Capture migration-seeded rows (singleton config like import_settings
+    / ml_settings) once, so the per-test truncation can restore them."""
+    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()
+    return snap
+
+
 @pytest.fixture(autouse=True)
-def _truncate_after_integration(request):
+def _reset_db_after_integration(request, _seed_snapshot):
     """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. Truncate every model table after each
-    integration-marked test. Scoped to the integration marker so the
-    DB-less fast unit job never connects here.
+    would leak between tests. After each integration-marked test, truncate
+    every model table, then restore the migration-seeded rows. Scoped to
+    the integration marker so the DB-less fast unit job never connects.
     """
     yield
     if request.node.get_closest_marker("integration") is None:
@@ -97,5 +116,24 @@ def _truncate_after_integration(request):
             conn.exec_driver_sql(
                 f"TRUNCATE {tables} RESTART IDENTITY CASCADE"
             )
+            for t in Base.metadata.sorted_tables:
+                rows = _seed_snapshot.get(t.name)
+                if rows:
+                    conn.execute(t.insert(), rows)
     finally:
         eng.dispose()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _dispose_app_engine(request):
+    """The app's shared async engine (extensions.get_engine) is a process
+    singleton, but pytest-asyncio gives each test a fresh event loop;
+    asyncpg connections are loop-bound, so reuse across tests raises
+    'attached to a different loop'. Dispose after each integration test so
+    the next test rebuilds the engine on its own loop.
+    """
+    yield
+    if request.node.get_closest_marker("integration") is None:
+        return
+    from backend.app import extensions
+    await extensions.dispose_engine()