https://github.com/krylosov-aa/context-async-sqlalchemy/issues/8

Author: krylosov-aa

exmaples/fastapi_example/tests/transactional/conftest.py

@@ -14,7 +14,7 @@
 
 import pytest_asyncio
 
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import async_sessionmaker, AsyncSession
 
 from exmaples.fastapi_example.database import master
 from context_async_sqlalchemy import (
@@ -24,6 +24,18 @@
 )
 
 
+@pytest_asyncio.fixture
+async def db_session_test(
+    session_maker_test: async_sessionmaker[AsyncSession],
+) -> AsyncGenerator[AsyncSession]:
+    """The session that is used inside the test"""
+    async with session_maker_test() as session:
+        try:
+            yield session
+        finally:
+            await session.rollback()
+
+
 @pytest_asyncio.fixture(autouse=True)
 async def db_session_override(
     db_session_test: AsyncSession,
@@ -34,7 +46,15 @@ async def db_session_override(
         if it already exists.
     """
     token = init_db_session_ctx()
-    put_db_session_to_context(master.context_key, db_session_test)
+
+    # Here we create a new session with save point behavior.
+    # This means that committing within the application will save and
+    #   release the save point, rather than committing the entire transaction.
+    conn = await db_session_test.connection()
+    new_session = AsyncSession(
+        bind=conn, join_transaction_mode="create_savepoint"
+    )
+    put_db_session_to_context(master.context_key, new_session)
     try:
         yield
     finally:

exmaples/fastapi_example/tests/transactional/test_routes.py

@@ -36,3 +36,25 @@ async def test_example_with_db_session(
     result = await db_session_test.execute(select(ExampleTable))
     row = result.scalar_one()
     assert row.text == "example_with_db_session"
+
+
+@pytest.mark.asyncio
+async def test_example_with_db_session_and_atomic(
+    client: AsyncClient,
+    db_session_test: AsyncSession,
+) -> None:
+    """
+    Since the handler involves manual session management,
+        such a handler should only be tested in non-transactional tests.
+    """
+    # Act
+    response = await client.post(
+        "/example_with_db_session_and_atomic",
+    )
+
+    # Assert
+    assert response.status_code == HTTPStatus.OK
+
+    result = await db_session_test.execute(select(ExampleTable))
+    row = result.scalar_one()
+    assert row.text == "example_with_db_session_and_atomic"