reuse db connection in tests

Author: lesnik512

tests/conftest.py

@@ -6,7 +6,7 @@
 import pytest
 from asgi_lifespan import LifespanManager
 from httpx import ASGITransport, AsyncClient
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import AsyncConnection, AsyncSession
 
 from app import ioc
 from app.application import build_app
@@ -33,18 +33,27 @@ def di_container(app: fastapi.FastAPI) -> modern_di.Container:
     return modern_di_fastapi.fetch_di_container(app)
 
 
-@pytest.fixture(autouse=True)
-async def db_session(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncSession]:
+@pytest.fixture(scope="session")
+async def db_connection(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncConnection]:
     engine = await ioc.Dependencies.database_engine.async_resolve(di_container)
-    connection = await engine.connect()
-    transaction = await connection.begin()
-    await connection.begin_nested()
-    ioc.Dependencies.database_engine.override(connection, di_container)
-
+    connection: typing.Final = await engine.connect()
     try:
-        yield AsyncSession(connection, expire_on_commit=False, autoflush=False)
+        yield connection
     finally:
-        if connection.in_transaction():
-            await transaction.rollback()
         await connection.close()
         await engine.dispose()
+
+
+@pytest.fixture(autouse=True)
+async def db_session(
+    db_connection: AsyncConnection, di_container: modern_di.Container
+) -> typing.AsyncIterator[AsyncSession]:
+    transaction = await db_connection.begin()
+    await db_connection.begin_nested()
+    ioc.Dependencies.database_engine.override(db_connection, di_container)
+
+    try:
+        yield AsyncSession(db_connection, expire_on_commit=False, autoflush=False)
+    finally:
+        if db_connection.in_transaction():
+            await transaction.rollback()