Python Contract Testing — Deep Dive

Consumer-side Pact tests in Python

The consumer writes tests that describe the interactions it expects from a provider:

# tests/contract/test_user_service_consumer.py
import pytest
from pact import Consumer, Provider

from myapp.clients.user_client import UserClient

@pytest.fixture(scope="module")
def pact():
    """Set up Pact mock provider."""
    pact = Consumer("NotificationService").has_pact_with(
        Provider("UserService"),
        pact_dir="./pacts",
    )
    pact.start_service()
    yield pact
    pact.stop_service()
    pact.verify()

def test_get_user_for_notification(pact):
    """NotificationService needs user email and name."""
    expected_body = {
        "id": 1,
        "name": "Alice Johnson",
        "email": "alice@example.com",
    }
    
    (pact
     .given("a user with ID 1 exists")
     .upon_receiving("a request for user 1")
     .with_request("GET", "/api/users/1")
     .will_respond_with(200, body=expected_body))
    
    # Use the actual client code against the Pact mock
    client = UserClient(base_url=pact.uri)
    user = client.get_user(1)
    
    assert user.name == "Alice Johnson"
    assert user.email == "alice@example.com"

def test_user_not_found(pact):
    """NotificationService handles missing users gracefully."""
    (pact
     .given("no user with ID 999 exists")
     .upon_receiving("a request for non-existent user 999")
     .with_request("GET", "/api/users/999")
     .will_respond_with(404, body={"error": "User not found"}))
    
    client = UserClient(base_url=pact.uri)
    user = client.get_user(999)
    
    assert user is None

This generates a Pact file in ./pacts/ — a JSON document describing the expected interactions. The consumer tests verify that the consumer code correctly handles the expected responses.

Provider-side verification

The provider runs consumer contracts against its real implementation:

# tests/contract/test_user_service_provider.py
import pytest
from pact import Verifier

from user_service.app import create_app
from user_service.database import seed_test_data

@pytest.fixture(scope="module")
def provider_app():
    """Start the actual provider application."""
    app = create_app(config="testing")
    with app.app_context():
        seed_test_data()
    return app

def test_verify_pacts(provider_app):
    """Verify all consumer contracts against real provider."""
    verifier = Verifier(
        provider="UserService",
        provider_base_url="http://localhost:5000",
    )
    
    # Provider states map to data setup
    verifier.provider_states_setup_url = "http://localhost:5000/_pact/setup"
    
    output, _ = verifier.verify_pacts(
        "./pacts/notificationservice-userservice.json",
        verbose=True,
    )
    assert output == 0

Provider states are critical. The given("a user with ID 1 exists") clause tells the provider to set up specific data before running the interaction:

# user_service/routes/pact_setup.py
from flask import Blueprint, request, jsonify
from user_service.database import db, User

pact_bp = Blueprint("pact", __name__)

@pact_bp.route("/_pact/setup", methods=["POST"])
def provider_state_setup():
    """Set up data for Pact provider states."""
    state = request.json.get("state", "")
    
    if state == "a user with ID 1 exists":
        db.session.query(User).delete()
        db.session.add(User(id=1, name="Alice Johnson", email="alice@example.com"))
        db.session.commit()
    elif state == "no user with ID 999 exists":
        db.session.query(User).filter_by(id=999).delete()
        db.session.commit()
    
    return jsonify({"status": "ok"})

Using Pact Broker for coordination

The Pact Broker stores contracts centrally and tracks compatibility between service versions:

# Publish consumer pact after consumer tests pass
pact-broker publish ./pacts \
  --consumer-app-version=$(git rev-parse --short HEAD) \
  --branch=$(git branch --show-current) \
  --broker-base-url=https://pact-broker.internal

# Verify provider against all consumer pacts
pact-verifier \
  --provider-base-url=http://localhost:5000 \
  --pact-broker-base-url=https://pact-broker.internal \
  --provider=UserService \
  --provider-app-version=$(git rev-parse --short HEAD) \
  --publish-verification-results

# Check deployment safety
pact-broker can-i-deploy \
  --pacticipant=UserService \
  --version=$(git rev-parse --short HEAD) \
  --to-environment=production

The can-i-deploy command is the key CI gate. It checks the broker’s compatibility matrix: “Given the latest consumer contracts, does this provider version satisfy them all?” If not, deployment is blocked.

Event-driven contract testing

For services communicating via message queues (Kafka, RabbitMQ), Pact supports message-level contracts:

# Consumer: defines expected message format
def test_order_created_event_consumer(pact):
    """InventoryService expects order events with specific fields."""
    expected_message = {
        "order_id": "ord-123",
        "items": [
            {"sku": "WIDGET-001", "quantity": 5}
        ],
        "created_at": "2026-03-28T10:00:00Z",
    }
    
    (pact
     .given("an order is created")
     .expects_to_receive("an OrderCreated event")
     .with_content(expected_message)
     .with_metadata({"content-type": "application/json"}))

# Provider: verifies it produces messages matching the contract
def test_order_created_event_provider():
    """OrderService produces events matching consumer expectations."""
    verifier = MessageVerifier(
        provider="OrderService",
        provider_base_url="http://localhost:5001",
    )
    
    verifier.verify_with_broker(
        broker_url="https://pact-broker.internal",
        publish_version=git_version(),
    )

The provider verification calls an endpoint that triggers message creation and returns the message content. Pact then checks the content matches consumer expectations.

Schema-based contract testing with Schemathesis

For OpenAPI/Swagger-documented APIs, Schemathesis provides contract testing without writing explicit consumer tests:

# Generate tests from OpenAPI spec
import schemathesis

schema = schemathesis.from_uri("http://localhost:5000/openapi.json")

@schema.parametrize()
def test_api_contract(case):
    """Every endpoint responds according to its OpenAPI spec."""
    response = case.call()
    case.validate_response(response)

Schemathesis automatically generates requests for every documented endpoint, testing edge cases like missing required fields, wrong types, and boundary values. It verifies responses match the documented schema.

CI pipeline architecture

A complete contract testing pipeline coordinates consumer and provider builds:

# Consumer CI (.github/workflows/consumer-ci.yml)
jobs:
  test-and-publish-pact:
    steps:
      - name: Run consumer tests (generates pact)
        run: pytest tests/contract/ -v
      - name: Publish pact to broker
        run: |
          pact-broker publish ./pacts \
            --consumer-app-version=${{ github.sha }} \
            --branch=${{ github.ref_name }} \
            --broker-base-url=${{ secrets.PACT_BROKER_URL }}
      - name: Trigger provider verification
        run: |
          curl -X POST "${{ secrets.PACT_BROKER_URL }}/webhooks" \
            -d '{"events": [{"name": "contract_content_changed"}]}'

# Provider CI (.github/workflows/provider-ci.yml)
jobs:
  verify-contracts:
    steps:
      - name: Verify all consumer contracts
        run: |
          pytest tests/contract/test_provider.py -v
      - name: Check deployment safety
        run: |
          pact-broker can-i-deploy \
            --pacticipant=UserService \
            --version=${{ github.sha }} \
            --to-environment=production

The webhook triggers provider verification whenever a consumer publishes a new contract. The can-i-deploy check gates deployment to ensure compatibility.

Tradeoffs and practical considerations

Contract testing adds infrastructure (broker, CI hooks) and requires both teams to participate. If the provider team doesn’t run verifications, contracts provide no value.

Start with your most critical integration points — the ones that have broken in production before. Expand to other service boundaries as the team builds familiarity with the workflow.

The biggest pitfall is over-specifying contracts. A contract that checks exact field values instead of types and structure becomes a brittle integration test. Keep contracts focused on shape and structure, not business logic.

One thing to remember: Contract testing’s power comes from decoupling — each team tests independently against a shared agreement, catching integration breaks at build time rather than deploy time. The broker is the connective tissue that makes this work at scale.