Contract Testing with Pact: Consumer-Driven API Contracts

What Is Contract Testing

Contract testing solves the hardest problem in microservices: how do you know that Service A's expectations about Service B's API are actually satisfied — without running both services together in a slow, fragile integration environment?

The answer is a contract: a recorded set of interactions that captures exactly what the consumer expects. The provider then verifies it can satisfy that contract independently. No live network call between them is needed.

This is fundamentally different from:

• Integration tests: Both services must be running simultaneously
• Schema tests: Check the provider schema but not the consumer's actual usage
• E2E tests: Require a full deployed environment

Contract tests run in milliseconds, never flake due to network issues, and catch breaking changes before they reach staging.

Consumer-Side Pact

The consumer writes tests that describe what they expect from the provider. Pact spins up a local mock server and records the interaction into a pact file (JSON).

# pip install pact-python
import pytest
from pact import Consumer, Provider, Like, Term
from myapp.clients import UserServiceClient

PACT_MOCK_HOST = 'localhost'
PACT_MOCK_PORT = 1234

@pytest.fixture(scope='session')
def pact():
    pact = Consumer('order-service').has_pact_with(
        Provider('user-service'),
        host_name=PACT_MOCK_HOST,
        port=PACT_MOCK_PORT,
        pact_dir='./pacts'
    )
    pact.start_service()
    yield pact
    pact.stop_service()

def test_get_existing_user(pact):
    expected_body = {
        'id': Like(42),
        'name': Like('Alice'),
        'email': Term(r'.+@.+', '[email protected]'),
    }
    (
        pact
        .given('user 42 exists')
        .upon_receiving('a GET request for user 42')
        .with_request('GET', '/users/42',
                      headers={'Accept': 'application/json'})
        .will_respond_with(200, body=expected_body)
    )
    with pact:
        client = UserServiceClient(base_url=f'http://{PACT_MOCK_HOST}:{PACT_MOCK_PORT}')
        user = client.get_user(42)
    assert user['id'] == 42

def test_get_missing_user_returns_404(pact):
    (
        pact
        .given('user 999 does not exist')
        .upon_receiving('a GET request for missing user')
        .with_request('GET', '/users/999')
        .will_respond_with(404, body={'error': Like('User not found')})
    )
    with pact:
        client = UserServiceClient(base_url=f'http://{PACT_MOCK_HOST}:{PACT_MOCK_PORT}')
        with pytest.raises(UserNotFoundError):
            client.get_user(999)

Pact Matchers

Pact matchers let you be flexible about values while being strict about structure:

Use Like() for fields that change between environments (IDs, timestamps). Use exact values for fields that are part of the contract (status enums, error codes).

Provider Verification

The provider reads the pact file and replays each interaction against its real implementation. For each interaction, the provider must:

• Set up the required state (e.g., 'user 42 exists')
• Handle the request
• Return a response that matches the pact

# Provider verification (pytest-pact or pact-verifier)
import pytest
from pact import Verifier

PROVIDER_URL = 'http://localhost:8000'

@pytest.fixture(scope='session')
def provider_app():
    # Start your actual provider app
    import subprocess
    proc = subprocess.Popen(['uvicorn', 'myapp:app', '--port', '8000'])
    yield
    proc.terminate()

def test_provider_satisfies_consumer_pacts(provider_app):
    verifier = Verifier(provider='user-service', provider_base_url=PROVIDER_URL)
    output, _ = verifier.verify_pacts(
        './pacts/order-service-user-service.json',
        provider_states_setup_url=f'{PROVIDER_URL}/_pact/provider-states'
    )
    assert output == 0  # 0 = all pacts satisfied

State Handlers

Provider state handlers create the test data required for each interaction:

# FastAPI provider state endpoint
from fastapi import FastAPI, Request

@app.post('/_pact/provider-states')
async def setup_provider_state(request: Request):
    body = await request.json()
    state = body['state']

    if state == 'user 42 exists':
        db.users.insert({'id': 42, 'name': 'Alice', 'email': '[email protected]'})
    elif state == 'user 999 does not exist':
        db.users.delete_where(id=999)  # ensure it doesn't exist

    return {'result': 'state setup complete'}

Pact Broker

The Pact Broker is a shared repository for pact files. It enables:

• Consumers publish pacts after each build
• Providers fetch and verify all consumer pacts
• can-i-deploy check before deployment

# Publish pacts to broker (from consumer CI)
pact-broker publish ./pacts \
  --broker-base-url https://broker.example.com \
  --consumer-app-version $GIT_SHA \
  --branch $GIT_BRANCH

# Check if it's safe to deploy (from any service CI)
pact-broker can-i-deploy \
  --pacticipant order-service \
  --version $GIT_SHA \
  --to-environment production

can-i-deploy

The can-i-deploy check is the key safety gate. Before deploying any service, it verifies that the version you want to deploy is compatible with all currently deployed versions of its dependencies. If user-service v2 breaks a pact that order-service v1.5 depends on, can-i-deploy fails and blocks the deployment.

CI/CD Integration

The full Pact workflow in CI:

# .github/workflows/consumer.yml
- name: Run consumer tests and publish pact
  run: |
    pytest tests/pact/
    pact-broker publish ./pacts \
      --broker-base-url $PACT_BROKER_URL \
      --consumer-app-version ${{ github.sha }}

# .github/workflows/provider.yml
- name: Verify pacts from broker
  run: |
    PACT_BROKER_URL=$PACT_BROKER_URL pytest tests/pact/provider_test.py

- name: Can-I-Deploy
  run: |
    pact-broker can-i-deploy \
      --pacticipant user-service \
      --version ${{ github.sha }} \
      --to-environment production

What Contract Testing Does Not Replace

Contract tests verify the API interface — status codes, body structure, required fields. They do not verify business logic (that the correct user is returned), performance, or security. Keep your integration and E2E tests for those concerns.