Implement shopping cart

test/conftest.py

@@ -3,34 +3,42 @@ from unittest import mock
 
 import pytest
 from fastapi.testclient import TestClient
-from sqlalchemy import StaticPool, create_engine
+from sqlalchemy import StaticPool, create_engine, event
 from sqlalchemy.orm import sessionmaker
 
 from allmende_payment_system.api.dependencies import get_session
 from allmende_payment_system.app import app
 from allmende_payment_system.models import Base
 
+engine = create_engine(
+    "sqlite:///:memory:",
+    connect_args={"check_same_thread": False},
+    poolclass=StaticPool,
+)
 
-def make_db():
-    engine = create_engine(
-        "sqlite:///:memory:",
-        connect_args={"check_same_thread": False},
-        poolclass=StaticPool,
-    )
-    Base.metadata.create_all(bind=engine)  # Create tables
-    return sessionmaker(autocommit=False, autoflush=False, bind=engine)
+# Create a single connection and an outer transaction which we can rollback
+# at the end of the test run. Individual tests will use nested transactions
+# (SAVEPOINTs) for isolation. This ensures the TestClient (app) and the
+# test fixture sessions see the same in-memory database state.
+connection = engine.connect()
+transaction = connection.begin()
+
+Base.metadata.create_all(bind=connection)
+
+# Bind sessions to the single connection so all sessions share the same DB
+TestSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=connection)
 
 
-def make_in_memory_session():
-    db = make_db()
-    session = db()
+def get_test_session():
+    """Dependency override for get_session"""
+    db = TestSessionLocal()
     try:
-        yield session
+        yield db
     finally:
-        session.close()
+        db.close()
 
 
-app.dependency_overrides[get_session] = make_in_memory_session
+app.dependency_overrides[get_session] = get_test_session
 
 
 @pytest.fixture(scope="session")
@@ -45,7 +53,25 @@ def unauthorized_client():
     return TestClient(app)
 
 
-@pytest.fixture
+@pytest.fixture(scope="function")
 def test_db():
-    db = make_db()
-    return db()
+    """Provides a database session for direct test usage"""
+    db = TestSessionLocal()
+
+    # Start a SAVEPOINT so test changes can be rolled back without
+    # closing the shared connection. Also restart the nested transaction
+    # when the session issues commits internally.
+    db.begin_nested()
+
+    @event.listens_for(db, "after_transaction_end")
+    def restart_savepoint(session, transaction):
+        # If the nested transaction ended, re-open it for continued isolation
+        if transaction.nested and not session.is_active:
+            session.begin_nested()
+
+    try:
+        yield db
+    finally:
+        # Rollback to the SAVEPOINT and close the session to clean up
+        db.rollback()
+        db.close()

test/test_database.py

@@ -11,7 +11,7 @@ def test_ensure_user(test_db):
 
     assert user.username == "test"
 
-    test_db.commit()
+    test_db.flush()
 
     assert test_db.scalar(select(func.count()).select_from(User)) == 1