fix(integration): lazy seed-snapshot (unblock unit job); isolate recover test from eager import

tests/conftest.py

@@ -77,35 +77,45 @@ 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
+# Migration-seeded baseline (singleton config like import_settings /
+# ml_settings), captured lazily at the FIRST integration test's setup —
+# when the CI integration job has just run `alembic upgrade head` so the DB
+# is pristine. Cached module-wide. Never populated in the DB-less unit job
+# because only integration-marked tests trigger the connection.
+_SEED_SNAPSHOT: dict[str, list[dict]] | None = None
 
 
 @pytest.fixture(autouse=True)
-def _reset_db_after_integration(request, _seed_snapshot):
+def _reset_db_after_integration(request):
     """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. 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.
+    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.
     """
+    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()
+        _SEED_SNAPSHOT = snap
+
     yield
-    if request.node.get_closest_marker("integration") is None:
+
+    if not is_integration:
         return
     tables = ", ".join(t.name for t in Base.metadata.sorted_tables)
     if not tables:
@@ -117,7 +127,7 @@ def _reset_db_after_integration(request, _seed_snapshot):
                 f"TRUNCATE {tables} RESTART IDENTITY CASCADE"
             )
             for t in Base.metadata.sorted_tables:
-                rows = _seed_snapshot.get(t.name)
+                rows = (_SEED_SNAPSHOT or {}).get(t.name)
                 if rows:
                     conn.execute(t.insert(), rows)
     finally: