Integration Testing: Test Real Dependencies, Not Mocks

Unit tests with mocks tell you that your code calls the right methods with the right arguments. Integration tests tell you that your code actually works. When your ORM generates invalid SQL, when your API client mishandles a 429 response, when your database query returns results in an unexpected order - unit tests with mocks pass. Integration tests catch the problem.

Why mocks lie

Consider this repository method and its mocked unit test.

# repository.py
class UserRepository:
    def __init__(self, db_session):
        self.db = db_session

    def find_active_users_by_name(self, name_prefix: str) -> list[User]:
        return (
            self.db.query(User)
            .filter(User.name.ilike(f"{name_prefix}%"))
            .filter(User.is_active == True)
            .order_by(User.created_at.desc())
            .all()
        )

# test_repository.py (mocked - what most teams write)
def test_find_active_users_by_name():
    mock_db = MagicMock()
    mock_query = MagicMock()
    mock_db.query.return_value = mock_query
    mock_query.filter.return_value = mock_query
    mock_query.order_by.return_value = mock_query
    mock_query.all.return_value = [User(name="Alice", is_active=True)]

    repo = UserRepository(mock_db)
    result = repo.find_active_users_by_name("Ali")

    assert len(result) == 1
    assert result[0].name == "Alice"
    # This test passes... but what does it actually verify?

This test verifies that .query(), .filter(), .order_by(), and .all() are called in some chain. It does not verify that the SQL is correct. It does not verify that ilike works as expected. It does not catch a typo in the column name. It does not verify the ordering.

Here is the integration test version.

# test_repository_integration.py
import pytest
from testcontainers.postgres import PostgresContainer

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

@pytest.fixture
def db_session(postgres):
    engine = create_engine(postgres.get_connection_url())
    Base.metadata.create_all(engine)
    session = Session(engine)
    yield session
    session.rollback()
    session.close()

@pytest.fixture
def seed_users(db_session):
    users = [
        User(name="Alice", is_active=True, created_at=datetime(2026, 1, 1)),
        User(name="Alicia", is_active=True, created_at=datetime(2026, 6, 1)),
        User(name="Albert", is_active=False, created_at=datetime(2026, 3, 1)),
        User(name="Bob", is_active=True, created_at=datetime(2026, 2, 1)),
    ]
    db_session.add_all(users)
    db_session.flush()
    return users

def test_find_active_users_by_name(db_session, seed_users):
    repo = UserRepository(db_session)
    result = repo.find_active_users_by_name("Ali")

    assert len(result) == 2
    assert result[0].name == "Alicia"  # Most recent first
    assert result[1].name == "Alice"
    # Albert is excluded (inactive), Bob is excluded (wrong prefix)

This test verifies the actual SQL against a real PostgreSQL database. It catches real bugs.

Setting up testcontainers

Testcontainers spins up real Docker containers for your tests. No shared test databases, no manual setup, no state leaking between test runs.

Python with testcontainers

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
from sqlalchemy import create_engine
from sqlalchemy.orm import Session

@pytest.fixture(scope="session")
def postgres_container():
    """One PostgreSQL container shared across all tests in the session."""
    with PostgresContainer("postgres:16") as container:
        yield container

@pytest.fixture(scope="session")
def redis_container():
    """One Redis container shared across all tests."""
    with RedisContainer("redis:7") as container:
        yield container

@pytest.fixture
def db_engine(postgres_container):
    engine = create_engine(postgres_container.get_connection_url())
    Base.metadata.create_all(engine)
    return engine

@pytest.fixture
def db_session(db_engine):
    """Each test gets its own session that rolls back after the test."""
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    yield session

    session.close()
    transaction.rollback()
    connection.close()

Node.js with testcontainers

// setup.js
import { PostgreSqlContainer } from "@testcontainers/postgresql";
import { Client } from "pg";

let container;
let connectionString;

beforeAll(async () => {
  container = await new PostgreSqlContainer("postgres:16").start();
  connectionString = container.getConnectionUri();

  // Run migrations
  const client = new Client({ connectionString });
  await client.connect();
  await client.query(`
    CREATE TABLE users (
      id SERIAL PRIMARY KEY,
      name VARCHAR(100) NOT NULL,
      email VARCHAR(255) UNIQUE NOT NULL,
      is_active BOOLEAN DEFAULT true,
      created_at TIMESTAMP DEFAULT NOW()
    )
  `);
  await client.end();
}, 30000); // Container startup can take time

afterAll(async () => {
  await container.stop();
});

// Export for use in tests
export function getConnectionString() {
  return connectionString;
}

// userRepository.test.js
import { getConnectionString } from "./setup";
import { UserRepository } from "../src/userRepository";
import { Client } from "pg";

describe("UserRepository", () => {
  let client;
  let repo;

  beforeEach(async () => {
    client = new Client({ connectionString: getConnectionString() });
    await client.connect();
    repo = new UserRepository(client);

    // Seed test data
    await client.query(`
      INSERT INTO users (name, email, is_active) VALUES
        ('Alice', 'alice@example.com', true),
        ('Bob', 'bob@example.com', true),
        ('Charlie', 'charlie@example.com', false)
    `);
  });

  afterEach(async () => {
    await client.query("DELETE FROM users");
    await client.end();
  });

  test("findActiveUsers returns only active users", async () => {
    const users = await repo.findActiveUsers();

    expect(users).toHaveLength(2);
    expect(users.map((u) => u.name).sort()).toEqual(["Alice", "Bob"]);
  });
});

Testing HTTP APIs

Integration tests for your API should test the full request/response cycle, including middleware, validation, serialization, and database interactions.

# test_api_integration.py
import pytest
from fastapi.testclient import TestClient
from app.main import create_app

@pytest.fixture
def app(db_session):
    """Create the FastAPI app with the test database session."""
    app = create_app()
    app.dependency_overrides[get_db] = lambda: db_session
    return app

@pytest.fixture
def client(app):
    return TestClient(app)

def test_create_order(client, db_session):
    # Create a user first
    user = User(name="Alice", email="alice@test.com")
    db_session.add(user)
    db_session.flush()

    # Create a product
    product = Product(name="Widget", price_cents=999, stock=10)
    db_session.add(product)
    db_session.flush()

    # Test the API endpoint
    response = client.post("/api/orders", json={
        "user_id": user.id,
        "items": [{"product_id": product.id, "quantity": 2}],
    })

    assert response.status_code == 201
    data = response.json()
    assert data["total_cents"] == 1998
    assert data["status"] == "pending"
    assert len(data["items"]) == 1

    # Verify side effects in the database
    db_session.expire_all()
    updated_product = db_session.get(Product, product.id)
    assert updated_product.stock == 8  # Decremented by 2


def test_create_order_insufficient_stock(client, db_session):
    user = User(name="Alice", email="alice@test.com")
    product = Product(name="Widget", price_cents=999, stock=1)
    db_session.add_all([user, product])
    db_session.flush()

    response = client.post("/api/orders", json={
        "user_id": user.id,
        "items": [{"product_id": product.id, "quantity": 5}],
    })

    assert response.status_code == 400
    assert "insufficient stock" in response.json()["detail"].lower()

    # Verify stock was NOT decremented
    db_session.expire_all()
    assert db_session.get(Product, product.id).stock == 1

Testing external services

For services you do not control (payment APIs, email providers), use a recorded response approach rather than mocks.

# Use responses library to intercept HTTP calls with realistic responses
import responses

@responses.activate
def test_payment_processing(client, db_session):
    # Record a realistic response from the payment provider
    responses.post(
        "https://api.stripe.com/v1/charges",
        json={
            "id": "ch_test_123",
            "amount": 1998,
            "currency": "usd",
            "status": "succeeded",
        },
        status=200,
    )

    # Create order through the full stack
    response = client.post("/api/orders", json={
        "user_id": 1,
        "items": [{"product_id": 1, "quantity": 2}],
        "payment_method": "pm_test_visa",
    })

    assert response.status_code == 201

    # Verify the correct request was sent to Stripe
    assert len(responses.calls) == 1
    request_body = responses.calls[0].request.body
    assert "amount=1998" in request_body


@responses.activate
def test_payment_failure_rolls_back_order(client, db_session):
    # Simulate payment provider failure
    responses.post(
        "https://api.stripe.com/v1/charges",
        json={"error": {"message": "Card declined"}},
        status=402,
    )

    response = client.post("/api/orders", json={
        "user_id": 1,
        "items": [{"product_id": 1, "quantity": 2}],
        "payment_method": "pm_test_declined",
    })

    assert response.status_code == 402

    # Verify no order was created
    orders = db_session.query(Order).all()
    assert len(orders) == 0

Keeping integration tests fast

Integration tests are slower than unit tests by nature, but they do not have to be slow.

Reuse containers across tests

# scope="session" means one container for the entire test run
@pytest.fixture(scope="session")
def postgres():
    with PostgresContainer("postgres:16") as pg:
        yield pg
# Container starts once, not once per test

Use transactions for test isolation

Instead of creating and dropping the database for each test, wrap each test in a transaction and roll back.

@pytest.fixture
def db_session(db_engine):
    connection = db_engine.connect()
    transaction = connection.begin()
    session = Session(bind=connection)

    yield session

    session.close()
    transaction.rollback()  # Undo all changes from this test
    connection.close()

Parallelize test execution

# pytest-xdist runs tests in parallel
pip install pytest-xdist
pytest -n auto tests/integration/

Each worker gets its own database container (or its own schema within a shared container).

@pytest.fixture(scope="session")
def db_schema(postgres, worker_id):
    """Each parallel worker gets its own schema."""
    schema_name = f"test_{worker_id}"
    engine = create_engine(postgres.get_connection_url())
    with engine.connect() as conn:
        conn.execute(text(f"CREATE SCHEMA IF NOT EXISTS {schema_name}"))
        conn.execute(text(f"SET search_path TO {schema_name}"))
        conn.commit()
    return schema_name

Separate fast and slow tests

# pytest.ini
[pytest]
markers =
    integration: marks tests as integration tests (deselect with '-m "not integration"')

@pytest.mark.integration
def test_database_query(db_session):
    ...

# Run only unit tests (fast)
pytest -m "not integration"

# Run everything (CI)
pytest

What to integration test

Not everything needs an integration test. Focus on the boundaries.

Do integration test:

• Database queries and ORM mappings
• API endpoints (request parsing, validation, response format)
• External service integrations (payment, email, auth)
• Message queue producers and consumers
• File storage operations
• Cache interactions

Keep as unit tests:

• Pure business logic (calculations, transformations)
• Input validation rules
• Data mapping functions
• Utility functions

Wrapping Up

Integration tests catch the bugs that unit tests with mocks miss: incorrect SQL, serialization mismatches, constraint violations, and unexpected behavior from real dependencies. Use testcontainers to spin up real databases and services in your test suite. Keep tests fast by reusing containers, using transaction rollback for isolation, and running tests in parallel. Focus integration testing on the boundaries of your system - the database queries, API endpoints, and external service calls where real behavior diverges from mocked behavior.