Integration Testing: Verifying the Seams Between Systems

Unit tests verify that individual functions work correctly in isolation. Integration tests verify that components work correctly together. The bugs that escape to production are almost always at the boundaries — the database query that works in tests but times out with real data, the API call that returns a different JSON shape than expected, the message queue consumer that silently drops messages with unexpected fields.

Integration tests are slower and harder to maintain than unit tests. This is the tradeoff: they catch the bugs that matter most.

---

What to Integration Test

Boundary	What Breaks	Test Strategy
Database	Queries, migrations, constraints	Test against real database (Testcontainers)
External APIs	Schema changes, rate limits, timeouts	Contract tests + recorded responses
Message queues	Serialization, dead letters, ordering	Test with real broker (in-memory)
File system	Permissions, paths, encoding	Temp directories, cleanup
Cache	Invalidation, TTL, serialization	Test with real Redis
Authentication	Token validation, session handling	Test middleware with real tokens

---

Database Integration Tests

# Using Testcontainers: real PostgreSQL, ephemeral
import pytest
from testcontainers.postgres import PostgresContainer

@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:16-alpine") as pg:
        # Run migrations
        engine = create_engine(pg.get_connection_url())
        run_migrations(engine)
        yield pg

@pytest.fixture
def db_session(postgres):
    engine = create_engine(postgres.get_connection_url())
    session = Session(engine)
    yield session
    session.rollback()  # Clean slate for next test
    session.close()

def test_create_order_persists_to_database(db_session):
    # Arrange
    user = create_user(db_session, email="test@example.com")
    product = create_product(db_session, name="Widget", price=9.99)

    # Act
    order = OrderService(db_session).create_order(
        user_id=user.id,
        items=[{"product_id": product.id, "quantity": 2}]
    )

    # Assert: verify it actually persisted
    saved = db_session.query(Order).filter_by(id=order.id).first()
    assert saved is not None
    assert saved.total == 19.98
    assert len(saved.items) == 1

def test_unique_email_constraint(db_session):
    create_user(db_session, email="duplicate@example.com")

    with pytest.raises(IntegrityError):
        create_user(db_session, email="duplicate@example.com")

---

API Contract Testing

# Verify your code handles the real API response format
# Record real responses, replay in tests

import responses
import json

@responses.activate
def test_payment_gateway_successful_charge():
    # Record from real API (or use VCR cassette)
    responses.add(
        responses.POST,
        "https://api.stripe.com/v1/charges",
        json={
            "id": "ch_test123",
            "status": "succeeded",
            "amount": 2000,
            "currency": "usd"
        },
        status=200
    )

    result = PaymentService().charge(amount=20.00, currency="usd", token="tok_visa")

    assert result.success is True
    assert result.charge_id == "ch_test123"

@responses.activate
def test_payment_gateway_handles_card_declined():
    responses.add(
        responses.POST,
        "https://api.stripe.com/v1/charges",
        json={
            "error": {
                "type": "card_error",
                "code": "card_declined",
                "message": "Your card was declined."
            }
        },
        status=402
    )

    result = PaymentService().charge(amount=20.00, currency="usd", token="tok_declined")

    assert result.success is False
    assert result.error_code == "card_declined"

---

Testcontainers: Real Dependencies in CI

Service	Container	Startup Time
PostgreSQL	postgres:16-alpine	~2 seconds
Redis	redis:7-alpine	~1 second
RabbitMQ	rabbitmq:3-management	~5 seconds
Elasticsearch	elasticsearch:8	~15 seconds
LocalStack (AWS)	localstack/localstack	~10 seconds
Kafka	confluentinc/cp-kafka	~15 seconds

# Multiple containers for full integration environment
@pytest.fixture(scope="session")
def integration_env():
    postgres = PostgresContainer("postgres:16-alpine")
    redis = RedisContainer("redis:7-alpine")

    postgres.start()
    redis.start()

    yield {
        "db_url": postgres.get_connection_url(),
        "redis_url": redis.get_connection_url(),
    }

    postgres.stop()
    redis.stop()

---

Test Data Management

Strategy	How	Best For
Transaction rollback	Wrap each test in a transaction, rollback after	Fast, clean isolation
Truncate tables	Delete all data between tests	When rollback is not possible
Factory functions	create_user(name="Test")	Readable, flexible test setup
Fixtures	Pre-loaded baseline data	Shared reference data (countries, categories)
Snapshots	Database dump/restore	Complex initial state

# Factory pattern for test data
class UserFactory:
    _counter = 0

    @classmethod
    def create(cls, db, **overrides):
        cls._counter += 1
        defaults = {
            "email": f"user_{cls._counter}@test.com",
            "name": f"Test User {cls._counter}",
            "role": "user",
        }
        defaults.update(overrides)
        user = User(**defaults)
        db.add(user)
        db.flush()
        return user

# Usage in tests
def test_admin_can_delete_user(db_session):
    admin = UserFactory.create(db_session, role="admin")
    target = UserFactory.create(db_session)

    result = UserService(db_session).delete_user(actor=admin, target_id=target.id)

    assert result.success is True

---

CI Pipeline Configuration

name: Integration Tests

on:
  pull_request:
    branches: [main]

jobs:
  integration:
    runs-on: ubuntu-latest
    services:
      postgres:
        image: postgres:16-alpine
        env:
          POSTGRES_DB: test
          POSTGRES_PASSWORD: test
        ports:
          - 5432:5432
        options: >-
          --health-cmd pg_isready
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5

      redis:
        image: redis:7-alpine
        ports:
          - 6379:6379

    steps:
      - uses: actions/checkout@v4
      - run: pip install -r requirements-test.txt
      - run: pytest tests/integration/ -v --timeout=30
        env:
          DATABASE_URL: postgresql://postgres:test@localhost:5432/test
          REDIS_URL: redis://localhost:6379

---

Implementation Checklist

• Write integration tests for every database query with complex logic (joins, aggregations)
• Use Testcontainers for real database/cache/queue dependencies in tests
• Test API contracts by recording and replaying real API responses
• Isolate each test: transaction rollback or table truncation between tests
• Use factory functions for test data creation — not raw SQL inserts
• Set timeouts on all integration tests (30 seconds max per test)
• Run integration tests on every PR in CI with service containers
• Test error paths: timeouts, connection failures, malformed responses
• Keep integration test suite under 5 minutes total runtime
• Separate unit and integration tests: tests/unit/ and tests/integration/