Python Integration Testing Patterns — Deep Dive

Database integration with transaction rollback

The transaction rollback pattern provides fast, isolated database testing. Each test runs inside a transaction that’s rolled back after the test, so no data persists:

# conftest.py
import pytest
from sqlalchemy import create_engine, event
from sqlalchemy.orm import sessionmaker

@pytest.fixture(scope="session")
def engine():
    return create_engine("postgresql://test:test@localhost:5432/testdb")

@pytest.fixture(scope="session")
def tables(engine):
    Base.metadata.create_all(engine)
    yield
    Base.metadata.drop_all(engine)

@pytest.fixture
def db_session(engine, tables):
    """Each test gets a transactional session that rolls back."""
    connection = engine.connect()
    transaction = connection.begin()
    session = sessionmaker(bind=connection)()
    
    # Handle nested transactions (SAVEPOINT)
    nested = connection.begin_nested()
    
    @event.listens_for(session, "after_transaction_end")
    def restart_savepoint(session, transaction):
        nonlocal nested
        if transaction.nested and not transaction._parent.nested:
            nested = connection.begin_nested()
    
    yield session
    
    session.close()
    transaction.rollback()
    connection.close()

The nested savepoint handler is crucial. If the code under test calls session.commit(), the savepoint is committed (not the outer transaction). The outer transaction still rolls back at the end, providing isolation while allowing the tested code to commit normally.

Testcontainers for reproducible infrastructure

Testcontainers spins up Docker containers for external services, ensuring tests use real implementations:

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer

@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:16-alpine") as pg:
        yield pg

@pytest.fixture(scope="session")
def redis():
    with RedisContainer("redis:7-alpine") as r:
        yield r

@pytest.fixture(scope="session")
def db_url(postgres):
    return postgres.get_connection_url()

@pytest.fixture(scope="session")
def redis_url(redis):
    host = redis.get_container_host_ip()
    port = redis.get_exposed_port(6379)
    return f"redis://{host}:{port}/0"

@pytest.fixture
def app(db_url, redis_url):
    """Application configured with test infrastructure."""
    return create_app(
        database_url=db_url,
        redis_url=redis_url,
        testing=True,
    )

Containers start once per session and are destroyed after all tests complete. The startup cost (5-15 seconds) is amortized across hundreds of tests.

API integration testing patterns

Test the full HTTP stack using framework test clients:

# tests/integration/test_user_api.py
import pytest
from fastapi.testclient import TestClient

class TestUserAPI:
    """Integration tests for the user management API."""
    
    def test_create_and_retrieve_user(self, client, db_session):
        """Full lifecycle: create user, verify storage, retrieve via API."""
        # Create
        response = client.post("/api/users", json={
            "name": "Alice Johnson",
            "email": "alice@example.com",
        })
        assert response.status_code == 201
        user_id = response.json()["id"]
        
        # Verify in database (not just API response)
        from models import User
        db_user = db_session.query(User).get(user_id)
        assert db_user is not None
        assert db_user.email == "alice@example.com"
        
        # Retrieve via API
        response = client.get(f"/api/users/{user_id}")
        assert response.status_code == 200
        assert response.json()["name"] == "Alice Johnson"
    
    def test_duplicate_email_rejected(self, client):
        """Database constraint prevents duplicate emails."""
        payload = {"name": "Alice", "email": "dupe@example.com"}
        client.post("/api/users", json=payload)
        
        response = client.post("/api/users", json=payload)
        assert response.status_code == 409
        assert "already exists" in response.json()["detail"]
    
    def test_authentication_required(self, client):
        """Protected endpoints reject unauthenticated requests."""
        response = client.get("/api/users/me")
        assert response.status_code == 401

    def test_pagination_with_real_data(self, client, make_user):
        """Pagination works correctly with actual database records."""
        for i in range(25):
            make_user(name=f"User {i}")
        
        page1 = client.get("/api/users?page=1&per_page=10")
        assert len(page1.json()["items"]) == 10
        assert page1.json()["total"] == 25
        
        page3 = client.get("/api/users?page=3&per_page=10")
        assert len(page3.json()["items"]) == 5

These tests catch issues that unit tests with mocked repositories miss: SQL errors, serialization problems, middleware interactions, and pagination edge cases.

Testing event-driven architectures

For systems using message queues, integration tests verify the full publish-consume cycle:

# tests/integration/test_order_events.py
import pytest
import json
import time

@pytest.fixture
def event_bus(redis_url):
    """In-memory event bus backed by Redis for testing."""
    from myapp.events import RedisPubSub
    bus = RedisPubSub(redis_url)
    yield bus
    bus.close()

class TestOrderEventFlow:
    
    def test_order_creation_publishes_event(self, client, event_bus):
        """Creating an order publishes an OrderCreated event."""
        received_events = []
        event_bus.subscribe("order.created", received_events.append)
        
        client.post("/api/orders", json={
            "items": [{"sku": "WIDGET-1", "quantity": 2}],
        })
        
        # Allow async processing
        time.sleep(0.5)
        
        assert len(received_events) == 1
        event = json.loads(received_events[0])
        assert event["type"] == "order.created"
        assert event["data"]["items"][0]["sku"] == "WIDGET-1"
    
    def test_inventory_updates_on_order(self, client, db_session, event_bus):
        """Inventory service processes OrderCreated events."""
        # Set initial stock
        from models import Inventory
        db_session.add(Inventory(sku="WIDGET-1", stock=10))
        db_session.commit()
        
        # Create order (triggers event → inventory handler)
        client.post("/api/orders", json={
            "items": [{"sku": "WIDGET-1", "quantity": 3}],
        })
        
        time.sleep(1.0)  # Wait for async processing
        
        inventory = db_session.query(Inventory).filter_by(sku="WIDGET-1").first()
        assert inventory.stock == 7

External API integration with record/replay

For third-party APIs, use VCR.py to record real responses and replay them in tests:

import pytest
import vcr

class TestPaymentIntegration:
    
    @vcr.use_cassette("tests/cassettes/stripe_charge_success.yaml")
    def test_successful_payment(self, payment_service):
        """First run records real Stripe API call; subsequent runs replay."""
        result = payment_service.charge(
            amount=2999,
            currency="usd",
            source="tok_visa",
        )
        assert result.status == "succeeded"
        assert result.amount == 2999
    
    @vcr.use_cassette("tests/cassettes/stripe_charge_declined.yaml")
    def test_declined_payment(self, payment_service):
        result = payment_service.charge(
            amount=2999,
            currency="usd",
            source="tok_chargeDeclined",
        )
        assert result.status == "failed"
        assert "declined" in result.error_message

Cassettes store the HTTP interaction, making tests fast and deterministic while ensuring they were validated against real APIs at recording time. Periodically re-record cassettes to catch API changes.

CI pipeline architecture for integration tests

Separate integration tests from unit tests in CI for optimal feedback:

# .github/workflows/test.yml
jobs:
  unit-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pytest tests/unit/ -x --tb=short -q
    # Fast feedback: ~30 seconds

  integration-tests:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16
        env:
          POSTGRES_DB: testdb
          POSTGRES_PASSWORD: test
        ports: ["5432:5432"]
      redis:
        image: redis:7
        ports: ["6379:6379"]
    steps:
      - uses: actions/checkout@v4
      - run: pytest tests/integration/ -x --tb=short -q
    # Thorough check: ~3-5 minutes

  e2e-tests:
    needs: [unit-tests, integration-tests]
    runs-on: ubuntu-latest
    steps:
      - run: pytest tests/e2e/ --tb=short
    # Full system: ~10-15 minutes

Unit tests run in parallel with integration tests. End-to-end tests only run if both pass. This gives developers fast feedback on the most common failures while still catching integration issues before merge.

Measuring integration test effectiveness

Track these metrics to keep your integration suite healthy:

• Unique bug detection: How many bugs do integration tests catch that unit tests miss? If the answer is “almost none,” the integration tests may be redundant.
• Execution time per test: Integration tests over 10 seconds each need investigation. Slow tests are usually doing too much or have inefficient setup.
• Flakiness rate: Integration tests are more flaky than unit tests due to external dependencies. Keep this under 1% by using deterministic data, proper timeouts, and container health checks.

One thing to remember: Integration tests are the most cost-effective investment for catching real-world bugs. Unit tests prove components work in isolation. Integration tests prove they work in production. The gap between those two is where most production bugs live.