Part 20: Integration, Contract & API Testing

Introduction — The Middle of the Pyramid

Unit tests prove that individual components work in isolation. But software systems are not collections of isolated components — they are networks of collaborating services, databases, message queues, and APIs. Integration tests verify that these components work correctly together.

The challenge is clear: integration tests are slower, more complex to set up, and more prone to flakiness than unit tests. But they catch an entire category of bugs that unit tests cannot — incorrect SQL queries, mismatched API contracts, serialisation errors, and network timeout handling.

Integration vs Unit — Key Differences

Types of Integration Tests

Component Integration

Tests a single service with its real dependencies (database, cache, filesystem) but without other services. This is the most common type in microservice architectures.

• Your API server + a real PostgreSQL database (via Testcontainers)
• Your worker + a real Redis cache
• Your file processor + a real filesystem

Service Integration

Tests the interaction between two or more services over the network. One service makes HTTP/gRPC calls to another, and both run during the test.

• Order Service calls Inventory Service via REST API
• Payment Gateway integration with your checkout service
• Authentication service issues JWT, API gateway validates it

System Integration

Tests the entire system end-to-end across all services. These overlap with E2E tests but focus on the backend interactions rather than the UI. Often run in staging environments.

flowchart TD
    subgraph Component["Component Integration"]
        A[Service A] --> DB[(Database)]
        A --> CACHE[(Redis Cache)]
    end
    subgraph Service["Service Integration"]
        B[Service A] -->|HTTP| C[Service B]
        C --> DB2[(Database)]
    end
    subgraph System["System Integration"]
        D[API Gateway] --> E[Auth Service]
        D --> F[Order Service]
        F --> G[Inventory Service]
        F --> H[Payment Service]
        G --> DB3[(Database)]
    end
                            

Testing with Real Dependencies

The revolution in integration testing came with Testcontainers — a library that spins up real Docker containers for databases, message brokers, and other services during test execution. No more in-memory fakes that behave differently from production.

Testcontainers — Docker-Based Testing

# Python — Testcontainers with PostgreSQL
import pytest
from testcontainers.postgres import PostgresContainer
import psycopg2

@pytest.fixture(scope="module")
def postgres_container():
    """Start a real PostgreSQL container for testing."""
    with PostgresContainer("postgres:16-alpine") as postgres:
        yield postgres

@pytest.fixture
def db_connection(postgres_container):
    """Create a database connection to the test container."""
    conn = psycopg2.connect(
        host=postgres_container.get_container_host_ip(),
        port=postgres_container.get_exposed_port(5432),
        user=postgres_container.username,
        password=postgres_container.password,
        database=postgres_container.dbname
    )
    # Setup: create tables
    with conn.cursor() as cur:
        cur.execute("""
            CREATE TABLE IF NOT EXISTS users (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                email VARCHAR(255) UNIQUE NOT NULL,
                created_at TIMESTAMP DEFAULT NOW()
            )
        """)
        conn.commit()
    yield conn
    # Teardown: close connection
    conn.close()

def test_insert_and_retrieve_user(db_connection):
    """Integration test: verify real SQL operations."""
    with db_connection.cursor() as cur:
        # Insert a user
        cur.execute(
            "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
            ("Alice", "alice@example.com")
        )
        user_id = cur.fetchone()[0]
        db_connection.commit()

        # Retrieve the user
        cur.execute("SELECT name, email FROM users WHERE id = %s", (user_id,))
        row = cur.fetchone()

    assert row[0] == "Alice"
    assert row[1] == "alice@example.com"

def test_unique_email_constraint(db_connection):
    """Integration test: database enforces uniqueness."""
    with db_connection.cursor() as cur:
        cur.execute(
            "INSERT INTO users (name, email) VALUES (%s, %s)",
            ("Bob", "unique@test.com")
        )
        db_connection.commit()

        # Attempting duplicate email should fail
        with pytest.raises(psycopg2.errors.UniqueViolation):
            cur.execute(
                "INSERT INTO users (name, email) VALUES (%s, %s)",
                ("Charlie", "unique@test.com")
            )
            db_connection.commit()
    db_connection.rollback()

// JavaScript — Testcontainers with PostgreSQL (using @testcontainers/postgresql)
const { PostgreSqlContainer } = require('@testcontainers/postgresql');
const { Client } = require('pg');

describe('User Repository Integration', () => {
    let container;
    let client;

    beforeAll(async () => {
        // Start a real PostgreSQL container
        container = await new PostgreSqlContainer('postgres:16-alpine').start();

        client = new Client({
            host: container.getHost(),
            port: container.getPort(),
            user: container.getUsername(),
            password: container.getPassword(),
            database: container.getDatabase()
        });
        await client.connect();

        // Create schema
        await client.query(`
            CREATE TABLE users (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100) NOT NULL,
                email VARCHAR(255) UNIQUE NOT NULL
            )
        `);
    }, 60000); // 60s timeout for container startup

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

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

    test('inserts and retrieves a user', async () => {
        await client.query(
            'INSERT INTO users (name, email) VALUES ($1, $2)',
            ['Alice', 'alice@test.com']
        );

        const result = await client.query('SELECT name, email FROM users WHERE email = $1', ['alice@test.com']);

        expect(result.rows[0]).toEqual({ name: 'Alice', email: 'alice@test.com' });
    });

    test('enforces unique email constraint', async () => {
        await client.query('INSERT INTO users (name, email) VALUES ($1, $2)', ['Bob', 'bob@test.com']);

        await expect(
            client.query('INSERT INTO users (name, email) VALUES ($1, $2)', ['Charlie', 'bob@test.com'])
        ).rejects.toThrow(/unique/i);
    });
});

In-Memory Alternatives

When Docker isn't available (CI constraints, local development speed), in-memory databases provide a compromise:

Setup & Teardown Strategies

• Transaction Rollback: Wrap each test in a transaction that rolls back. Fast but doesn't test commit behaviour.
• Truncate Tables: After each test, TRUNCATE all tables. Moderate speed, tests real commits.
• Recreate Schema: Drop and recreate the database for each test class. Slowest but cleanest.
• Unique Data: Use UUIDs or timestamps to make test data unique, avoiding conflicts between parallel tests.

API Testing

API tests validate that HTTP endpoints behave correctly — correct status codes, response shapes, error handling, headers, and content negotiation. They sit between integration tests and E2E tests.

SuperTest (Node.js) & httpx (Python)

// JavaScript — API testing with SuperTest
const request = require('supertest');
const app = require('./app'); // Express app instance

describe('GET /api/users', () => {
    test('returns 200 with array of users', async () => {
        const response = await request(app)
            .get('/api/users')
            .set('Accept', 'application/json');

        expect(response.status).toBe(200);
        expect(response.headers['content-type']).toMatch(/json/);
        expect(Array.isArray(response.body)).toBe(true);
    });

    test('supports pagination with limit and offset', async () => {
        const response = await request(app)
            .get('/api/users?limit=5&offset=10')
            .set('Accept', 'application/json');

        expect(response.status).toBe(200);
        expect(response.body.length).toBeLessThanOrEqual(5);
    });
});

describe('POST /api/users', () => {
    test('creates user and returns 201', async () => {
        const newUser = { name: 'Alice', email: 'alice@test.com' };

        const response = await request(app)
            .post('/api/users')
            .send(newUser)
            .set('Content-Type', 'application/json');

        expect(response.status).toBe(201);
        expect(response.body.id).toBeDefined();
        expect(response.body.name).toBe('Alice');
    });

    test('returns 400 for invalid email', async () => {
        const response = await request(app)
            .post('/api/users')
            .send({ name: 'Bob', email: 'not-an-email' })
            .set('Content-Type', 'application/json');

        expect(response.status).toBe(400);
        expect(response.body.error).toMatch(/email/i);
    });

    test('returns 409 for duplicate email', async () => {
        const user = { name: 'Charlie', email: 'duplicate@test.com' };
        await request(app).post('/api/users').send(user);

        const response = await request(app)
            .post('/api/users')
            .send(user);

        expect(response.status).toBe(409);
    });
});

# Python — API testing with httpx and pytest
import pytest
import httpx

BASE_URL = "http://localhost:8000"

@pytest.fixture(scope="module")
def client():
    """Create an HTTP client for API testing."""
    with httpx.Client(base_url=BASE_URL, timeout=10.0) as client:
        yield client

def test_get_users_returns_list(client):
    response = client.get("/api/users")

    assert response.status_code == 200
    assert response.headers["content-type"] == "application/json"
    data = response.json()
    assert isinstance(data, list)

def test_create_user_returns_201(client):
    payload = {"name": "Alice", "email": "alice@test.com"}
    response = client.post("/api/users", json=payload)

    assert response.status_code == 201
    body = response.json()
    assert body["name"] == "Alice"
    assert "id" in body

def test_create_user_invalid_email_returns_400(client):
    payload = {"name": "Bob", "email": "not-valid"}
    response = client.post("/api/users", json=payload)

    assert response.status_code == 400
    assert "email" in response.json().get("error", "").lower()

def test_get_nonexistent_user_returns_404(client):
    response = client.get("/api/users/99999")

    assert response.status_code == 404

Postman & Newman (CI Automation)

# Run Postman collection in CI with Newman
npm install -g newman

# Execute collection with environment variables
newman run collection.json \
  --environment staging.json \
  --reporters cli,junit \
  --reporter-junit-export results.xml

# Output:
# ┌─────────────────────────┬──────────┬──────────┐
# │                         │ executed │   failed │
# ├─────────────────────────┼──────────┼──────────┤
# │ iterations              │        1 │        0 │
# │ requests                │       12 │        0 │
# │ test-scripts            │       24 │        0 │
# │ assertions              │       36 │        0 │
# └─────────────────────────┴──────────┴──────────┘

Consumer-Driven Contract Testing

The Problem

In a microservices architecture, services evolve independently. Team A owns the User Service; Team B's Order Service consumes it. If Team A changes the API response shape — renames a field, removes a property, changes a type — Team B's service breaks in production. Integration tests might catch this, but only if both services are deployed to the same test environment simultaneously.

Solution: Consumer-Driven Contracts (Pact)

The Pact framework inverts the testing relationship. Instead of the provider defining the contract, consumers define what they expect. The provider then verifies it can fulfil all consumer expectations.

sequenceDiagram
    participant Consumer as Consumer Service
    participant PactBroker as Pact Broker
    participant Provider as Provider Service

    Consumer->>Consumer: Write consumer test
    Consumer->>Consumer: Generate pact file (JSON contract)
    Consumer->>PactBroker: Publish pact file
    Provider->>PactBroker: Fetch pact files from all consumers
    Provider->>Provider: Replay interactions against real provider
    Provider->>PactBroker: Publish verification results
    PactBroker->>Consumer: "Can I Deploy?" check passes/fails
                            

Consumer Test — Generating the Contract

// JavaScript — Pact consumer test (Order Service consuming User Service)
const { PactV3, MatchersV3 } = require('@pact-foundation/pact');
const { like, eachLike, integer, string } = MatchersV3;
const axios = require('axios');

const provider = new PactV3({
    consumer: 'OrderService',
    provider: 'UserService',
});

describe('User Service - Consumer Contract', () => {
    test('GET /api/users/:id returns user details', async () => {
        // Define expected interaction
        provider
            .given('a user with ID 1 exists')
            .uponReceiving('a request for user 1')
            .withRequest({
                method: 'GET',
                path: '/api/users/1',
                headers: { Accept: 'application/json' }
            })
            .willRespondWith({
                status: 200,
                headers: { 'Content-Type': 'application/json' },
                body: {
                    id: integer(1),
                    name: string('Alice'),
                    email: string('alice@example.com')
                }
            });

        // Execute test against mock provider
        await provider.executeTest(async (mockServer) => {
            const response = await axios.get(`${mockServer.url}/api/users/1`, {
                headers: { Accept: 'application/json' }
            });

            expect(response.status).toBe(200);
            expect(response.data.name).toBeDefined();
            expect(response.data.email).toBeDefined();
        });
        // Pact file (contract) is automatically generated
    });
});

Provider Verification — Replaying the Contract

// JavaScript — Pact provider verification (User Service)
const { Verifier } = require('@pact-foundation/pact');
const app = require('./app'); // The real User Service

describe('User Service - Provider Verification', () => {
    let server;

    beforeAll((done) => {
        server = app.listen(3001, done);
    });

    afterAll((done) => {
        server.close(done);
    });

    test('verifies all consumer contracts', async () => {
        const verifier = new Verifier({
            providerBaseUrl: 'http://localhost:3001',
            provider: 'UserService',
            pactBrokerUrl: 'https://your-org.pactflow.io',
            pactBrokerToken: process.env.PACT_BROKER_TOKEN,
            publishVerificationResult: true,
            providerVersion: process.env.GIT_COMMIT,
            stateHandlers: {
                'a user with ID 1 exists': async () => {
                    // Setup test state — ensure user 1 exists in DB
                    await db.users.create({ id: 1, name: 'Alice', email: 'alice@example.com' });
                }
            }
        });

        await verifier.verifyProvider();
    });
});

Schema Validation

Schema validation catches breaking changes at build time by comparing API responses against a formal schema definition. This is lighter than full contract testing but catches common issues.

# Python — JSON Schema validation in tests
import jsonschema
import json

USER_SCHEMA = {
    "type": "object",
    "required": ["id", "name", "email", "createdAt"],
    "properties": {
        "id": {"type": "integer"},
        "name": {"type": "string", "minLength": 1},
        "email": {"type": "string", "format": "email"},
        "createdAt": {"type": "string", "format": "date-time"},
        "role": {"type": "string", "enum": ["admin", "user", "moderator"]}
    },
    "additionalProperties": False
}

def validate_user_response(response_body):
    """Validate API response against schema."""
    jsonschema.validate(instance=response_body, schema=USER_SCHEMA)

def test_user_api_matches_schema():
    # Simulate API response
    api_response = {
        "id": 1,
        "name": "Alice",
        "email": "alice@example.com",
        "createdAt": "2026-05-13T10:30:00Z",
        "role": "admin"
    }

    # This will raise ValidationError if schema doesn't match
    validate_user_response(api_response)

def test_schema_rejects_missing_field():
    """Breaking change: removing 'email' field would fail validation."""
    import pytest
    broken_response = {"id": 1, "name": "Alice", "createdAt": "2026-05-13T10:30:00Z"}

    with pytest.raises(jsonschema.ValidationError):
        validate_user_response(broken_response)

# OpenAPI schema validation in CI
# Validate that your server responses match your OpenAPI spec
npx @schemathesis/cli run \
  --url http://localhost:3000/openapi.json \
  --checks all \
  --hypothesis-max-examples 100

# Output: Tests all endpoints with generated inputs,
# verifies responses match declared schemas

Service Virtualisation

When external services are unavailable, slow, or expensive to call (payment gateways, third-party APIs, partner systems), service virtualisation provides a controllable stand-in that mimics the real service's behaviour.

flowchart LR
    subgraph Test["Test Environment"]
        SUT[System Under Test]
        WM[WireMock / MockServer]
    end
    subgraph Prod["Production"]
        REAL[Real External API]
    end
    SUT -->|"In tests"| WM
    SUT -.->|"In production"| REAL
    WM -.- NOTE["Controllable responses
Simulated delays
Error injection"]
                            

// JavaScript — Using nock for HTTP interception
const nock = require('nock');
const axios = require('axios');

// Function under test
async function getWeather(city) {
    const response = await axios.get(`https://api.weather.com/current?city=${city}`);
    return {
        temperature: response.data.temp,
        description: response.data.weather
    };
}

test('getWeather extracts temperature and description', async () => {
    // Virtualise the external weather API
    nock('https://api.weather.com')
        .get('/current')
        .query({ city: 'London' })
        .reply(200, {
            temp: 12,
            weather: 'Cloudy',
            humidity: 78,
            wind_speed: 15
        });

    const result = await getWeather('London');

    expect(result).toEqual({ temperature: 12, description: 'Cloudy' });
});

test('getWeather handles API timeout', async () => {
    nock('https://api.weather.com')
        .get('/current')
        .query({ city: 'Paris' })
        .delayConnection(5000) // Simulate 5-second delay
        .reply(200, { temp: 18, weather: 'Sunny' });

    await expect(getWeather('Paris')).rejects.toThrow();
});

Testing Asynchronous Integrations

Many modern systems communicate via message queues (Kafka, RabbitMQ, SQS) or event streams. Testing these requires different approaches than synchronous HTTP.

# Python — Testing Kafka consumer with testcontainers
import pytest
import json
from testcontainers.kafka import KafkaContainer
from kafka import KafkaProducer, KafkaConsumer
import time

@pytest.fixture(scope="module")
def kafka_container():
    with KafkaContainer("confluentinc/cp-kafka:7.5.0") as kafka:
        yield kafka

def test_message_processing(kafka_container):
    """Test that our consumer correctly processes messages."""
    bootstrap_servers = kafka_container.get_bootstrap_server()
    topic = "user-events"

    # Produce a message
    producer = KafkaProducer(
        bootstrap_servers=bootstrap_servers,
        value_serializer=lambda v: json.dumps(v).encode()
    )
    event = {"type": "user_created", "user_id": 42, "name": "Alice"}
    producer.send(topic, value=event)
    producer.flush()

    # Consume and verify
    consumer = KafkaConsumer(
        topic,
        bootstrap_servers=bootstrap_servers,
        auto_offset_reset="earliest",
        value_deserializer=lambda v: json.loads(v.decode()),
        consumer_timeout_ms=10000
    )

    messages = []
    for message in consumer:
        messages.append(message.value)
        break  # Only need first message

    assert len(messages) == 1
    assert messages[0]["type"] == "user_created"
    assert messages[0]["user_id"] == 42

    consumer.close()
    producer.close()

Async Contract Testing

Pact supports message-based contract testing. The consumer defines what message shape it expects, and the provider verifies it can produce messages matching that shape.

// JavaScript — Pact message consumer test
const { MessageConsumerPact, MatchersV3 } = require('@pact-foundation/pact');
const { like, string, integer } = MatchersV3;

const messagePact = new MessageConsumerPact({
    consumer: 'NotificationService',
    provider: 'OrderService',
});

describe('Order Created Event Contract', () => {
    test('processes order.created event', () => {
        return messagePact
            .given('an order has been placed')
            .expectsToReceive('an order created event')
            .withContent({
                eventType: string('order.created'),
                orderId: integer(12345),
                userId: integer(1),
                totalAmount: like(99.99),
                items: [{ productId: integer(1), quantity: integer(2) }]
            })
            .verify(async (message) => {
                // Your message handler processes this message
                const result = processOrderEvent(message);
                expect(result.notificationSent).toBe(true);
            });
    });
});

Database Integration Testing

Migration Testing

Database schema migrations are one of the riskiest deployment activities. Testing them ensures they can be applied (and rolled back) safely.

# Run migrations forward and backward to verify reversibility
# Using Flyway (Java ecosystem)
flyway -url=jdbc:postgresql://localhost:5432/testdb migrate
flyway -url=jdbc:postgresql://localhost:5432/testdb undo

# Using Alembic (Python/SQLAlchemy)
alembic upgrade head      # Apply all migrations
alembic downgrade -1      # Rollback one step
alembic upgrade head      # Re-apply (idempotent check)

# Python — Migration test pattern
import subprocess
import pytest

def test_migrations_apply_cleanly():
    """Verify all migrations can be applied to an empty database."""
    result = subprocess.run(
        ["alembic", "upgrade", "head"],
        capture_output=True, text=True
    )
    assert result.returncode == 0, f"Migration failed: {result.stderr}"

def test_migrations_are_reversible():
    """Verify migrations can be rolled back."""
    # First apply all
    subprocess.run(["alembic", "upgrade", "head"], check=True)

    # Then rollback all
    result = subprocess.run(
        ["alembic", "downgrade", "base"],
        capture_output=True, text=True
    )
    assert result.returncode == 0, f"Rollback failed: {result.stderr}"

def test_migrations_are_idempotent():
    """Applying migrations twice should not error."""
    subprocess.run(["alembic", "upgrade", "head"], check=True)
    result = subprocess.run(
        ["alembic", "upgrade", "head"],
        capture_output=True, text=True
    )
    assert result.returncode == 0

Repository Pattern Testing

// JavaScript — Testing a repository with real database
const { PostgreSqlContainer } = require('@testcontainers/postgresql');
const { Pool } = require('pg');

class UserRepository {
    constructor(pool) {
        this.pool = pool;
    }

    async create(user) {
        const result = await this.pool.query(
            'INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *',
            [user.name, user.email]
        );
        return result.rows[0];
    }

    async findByEmail(email) {
        const result = await this.pool.query(
            'SELECT * FROM users WHERE email = $1',
            [email]
        );
        return result.rows[0] || null;
    }
}

describe('UserRepository Integration', () => {
    let container, pool, repo;

    beforeAll(async () => {
        container = await new PostgreSqlContainer().start();
        pool = new Pool({ connectionString: container.getConnectionUri() });
        await pool.query(`
            CREATE TABLE users (
                id SERIAL PRIMARY KEY,
                name VARCHAR(100),
                email VARCHAR(255) UNIQUE
            )
        `);
        repo = new UserRepository(pool);
    }, 60000);

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

    afterEach(async () => {
        await pool.query('DELETE FROM users');
    });

    test('create returns user with generated id', async () => {
        const user = await repo.create({ name: 'Alice', email: 'alice@test.com' });
        expect(user.id).toBeDefined();
        expect(user.name).toBe('Alice');
    });

    test('findByEmail returns null for nonexistent user', async () => {
        const user = await repo.findByEmail('nobody@test.com');
        expect(user).toBeNull();
    });

    test('findByEmail returns existing user', async () => {
        await repo.create({ name: 'Bob', email: 'bob@test.com' });
        const found = await repo.findByEmail('bob@test.com');
        expect(found.name).toBe('Bob');
    });
});

Integration Test Strategies

The Testing Honeycomb (Spotify Model)

Spotify's engineering team proposed the Testing Honeycomb as an alternative to the traditional pyramid for microservice architectures:

The honeycomb argues that in microservice architectures where services are small and well-defined, integration tests provide the best return on investment. Unit tests often test trivial logic, while integration tests verify the actual behaviour that users care about.

Speed vs Confidence Tradeoffs

Exercises

Conclusion & Next Steps

Integration tests bridge the gap between isolated unit tests and full system tests. They verify that real components — databases, HTTP APIs, message queues — work together correctly. Consumer-driven contracts (Pact) solve the coordination problem in microservice architectures by letting consumers define expectations independently of providers.

The key decisions are: how much to invest in each test layer, when to use real dependencies vs service virtualisation, and how to keep integration tests fast enough to run in CI. There is no universal answer — the right balance depends on your architecture, team size, and deployment cadence.

In the next article, we complete the testing pyramid by exploring end-to-end testing with modern frameworks like Playwright and Cypress, visual regression testing, and strategies for managing test flakiness at scale.