3.5Transactions: The Safety Net

Transactions ensure that test data changes are atomic (all or nothing) and can be rolled back.

Transaction for Test Setup

-- Everything succeeds or nothing does
BEGIN;

INSERT INTO users (id, name, email, role)
VALUES ('test-uuid-002', 'Transaction User', 'txn@test.com', 'viewer');

INSERT INTO orders (id, user_id, status, total)
VALUES ('order-txn', 'test-uuid-002', 'pending', 50.00);

-- If either INSERT fails, ROLLBACK undoes everything
COMMIT;

Transaction-Based Test Isolation

The most powerful pattern for test data management: wrap each test in a transaction and roll it back afterward. The database returns to its original state after every test.

@pytest.fixture
def db_transaction(db):
    """Wrap each test in a transaction that rolls back."""
    db.autocommit = False
    yield db
    db.rollback()  # All changes are undone after each test

def test_user_creation(db_transaction, api_client):
    api_client.post("/users", json={"name": "Alice", "email": "alice@test.com"})

    cursor = db_transaction.cursor()
    cursor.execute("SELECT * FROM users WHERE email = 'alice@test.com'")
    assert cursor.fetchone() is not None
    # After the test, db.rollback() removes the user automatically

def test_another_test(db_transaction):
    # This test starts with a clean database — no Alice from the previous test
    cursor = db_transaction.cursor()
    cursor.execute("SELECT * FROM users WHERE email = 'alice@test.com'")
    assert cursor.fetchone() is None  # Alice does not exist

Transaction Limitations

Scenario	Transaction Rollback Works?
Test creates data via direct SQL	Yes
Test creates data via API call to same DB	Yes (if API uses same connection)
Test creates data via API call to separate service	No (different connection/transaction)
Test modifies external systems (S3, email)	No (cannot rollback external side effects)

For API tests where rollback is not possible, use cleanup fixtures instead:

@pytest.fixture
def create_user(api_client):
    created_ids = []

    def _create(**kwargs):
        r = api_client.post("/users", json=kwargs)
        user = r.json()
        created_ids.append(user["id"])
        return user

    yield _create

    for uid in created_ids:
        api_client.delete(f"/users/{uid}")