test: prove db_session rollback with one self-contained test

Author: lesnik512

tests/conftest.py

@@ -1,3 +1,4 @@
+import contextlib
 import typing
 
 import litestar
@@ -38,8 +39,8 @@ async def di_container(app: litestar.Litestar) -> typing.AsyncIterator[modern_di
         await container.close_async()
 
 
-@pytest.fixture
-async def db_session(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncSession]:
+@contextlib.asynccontextmanager
+async def rolled_back_db_session(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncSession]:
     engine = create_database_engine()
     connection = await engine.connect()
     transaction = await connection.begin()
@@ -53,3 +54,9 @@ async def db_session(di_container: modern_di.Container) -> typing.AsyncIterator[
         await connection.close()
         await engine.dispose()
         di_container.reset_override(ioc.Database.database_engine)
+
+
+@pytest.fixture
+async def db_session(di_container: modern_di.Container) -> typing.AsyncIterator[AsyncSession]:
+    async with rolled_back_db_session(di_container) as session:
+        yield session

tests/test_main.py

@@ -11,6 +11,7 @@
 from app.database.tables import UsersTable
 from app.exceptions import PermissionDeniedError
 from app.settings import settings
+from tests.conftest import rolled_back_db_session
 from tests.factories import UserFactory
 
 
@@ -52,18 +53,24 @@ async def test_create_database_engine_reads_settings_and_can_be_disposed() -> No
         await close_database_engine(engine)
 
 
-async def test_db_session_insert_is_visible_within_test(db_session: AsyncSession) -> None:
-    user = UserFactory.build()
-    db_session.add(user)
-    await db_session.commit()
-
-    result = await db_session.scalars(sa.select(UsersTable))
-    assert len(result.all()) == 1
+async def test_db_session_teardown_rolls_back_committed_writes(di_container: modern_di.Container) -> None:
+    """INVARIANT: a commit inside a test only releases a savepoint, and teardown discards it.
 
+    The row is read back from an independent connection after teardown, so the check holds whatever runs before it.
+    """
+    user = UserFactory.build()
+    async with rolled_back_db_session(di_container) as session:
+        session.add(user)
+        await session.commit()
+        assert (await session.scalars(sa.select(UsersTable).where(UsersTable.id == user.id))).one() is user
 
-async def test_db_session_rolls_back_between_tests(db_session: AsyncSession) -> None:
-    result = await db_session.scalars(sa.select(UsersTable))
-    assert result.all() == []
+    engine = create_database_engine()
+    try:
+        async with engine.connect() as connection:
+            result = await connection.scalars(sa.select(UsersTable.id).where(UsersTable.id == user.id))
+            assert result.all() == []
+    finally:
+        await engine.dispose()
 
 
 async def test_di_resolved_session_shares_the_overridden_connection(