reuse db connection in tests

Author: lesnik512

pyproject.toml

@@ -76,7 +76,7 @@ pretty = true
 [tool.pytest.ini_options]
 addopts = "--cov=. --cov-report term-missing"
 asyncio_mode = "auto"
-asyncio_default_fixture_loop_scope = "function"
+asyncio_default_fixture_loop_scope = "session"
 
 [tool.coverage.report]
 exclude_also = [

tests/conftest.py

@@ -6,20 +6,20 @@
 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
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 async def app() -> typing.AsyncIterator[fastapi.FastAPI]:
     app_ = build_app()
     async with LifespanManager(app_):
         yield app_
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 async def client(app: fastapi.FastAPI) -> typing.AsyncIterator[AsyncClient]:
     async with AsyncClient(
         transport=ASGITransport(app=app),
@@ -28,23 +28,32 @@ async def client(app: fastapi.FastAPI) -> typing.AsyncIterator[AsyncClient]:
         yield client
 
 
-@pytest.fixture
+@pytest.fixture(scope="session")
 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()