📘 Contract Testing: API Contracts

🎯 Introduction

Welcome to this exciting tutorial on Contract Testing for API Contracts! 🎉 In this guide, we’ll explore how to ensure your APIs behave exactly as promised, preventing those dreaded integration failures.

You’ll discover how contract testing can transform your API development experience. Whether you’re building microservices 🏗️, RESTful APIs 🌐, or working with third-party integrations 🤝, understanding contract testing is essential for building reliable, maintainable systems.

By the end of this tutorial, you’ll feel confident implementing contract tests that catch breaking changes before they reach production! Let’s dive in! 🏊‍♂️

📚 Understanding Contract Testing

🤔 What is Contract Testing?

Contract testing is like a handshake agreement between services 🤝. Think of it as a legally binding document that ensures both the API provider and consumer stick to their promises!

In Python terms, contract testing verifies that:

✨ API providers deliver the exact response format promised
🚀 API consumers send requests in the expected format
🛡️ Changes don’t break existing integrations

💡 Why Use Contract Testing?

Here’s why developers love contract testing:

Early Detection 🔍: Catch breaking changes during development
Independent Testing 💻: Test services without full integration setup
Clear Documentation 📖: Contracts serve as living API docs
Deployment Confidence 🔧: Deploy services independently with confidence

Real-world example: Imagine an e-commerce system 🛒. With contract testing, you can ensure the payment service API always returns the expected response format, preventing checkout failures!

🔧 Basic Syntax and Usage

📝 Simple Example with Pact

Let’s start with a friendly example using Pact, a popular contract testing framework:

# 👋 Hello, Contract Testing!
from pact import Consumer, Provider

# 🎨 Creating a simple contract
pact = Consumer('OrderService').has_pact_with(
    Provider('PaymentService'),
    host_name='localhost',
    port=1234
)

# 📝 Define the expected interaction
(pact
 .given('a valid payment request')  # 🎯 Provider state
 .upon_receiving('a payment request')  # 📨 Request description
 .with_request('POST', '/payments',  # 🔧 Request details
               headers={'Content-Type': 'application/json'},
               body={'amount': 99.99, 'currency': 'USD'})
 .will_respond_with(200,  # ✅ Expected response
                   headers={'Content-Type': 'application/json'},
                   body={'status': 'approved', 'transaction_id': '12345'}))

💡 Explanation: We’re defining a contract between OrderService and PaymentService. The contract specifies exactly what request to expect and what response to return!

🎯 Common Patterns

Here are patterns you’ll use daily:

# 🏗️ Pattern 1: Consumer test
import requests
from pact import Consumer, Provider

def test_payment_approval():
    # 🎯 Set up the contract
    pact = Consumer('OrderService').has_pact_with(Provider('PaymentService'))
    
    with pact:
        # 📝 Define expected interaction
        pact.given('payment can be processed')
        pact.upon_receiving('a payment request')
        pact.with_request('POST', '/payments', 
                         body={'amount': 50.0})
        pact.will_respond_with(200, 
                              body={'status': 'approved'})
        
        # 🚀 Make the actual request
        response = requests.post(
            pact.uri + '/payments',
            json={'amount': 50.0}
        )
        
        # ✅ Verify the response
        assert response.status_code == 200
        assert response.json()['status'] == 'approved'

# 🎨 Pattern 2: Provider verification
from pact import Verifier

def test_payment_provider_honors_contract():
    verifier = Verifier(
        provider='PaymentService',
        provider_base_url='http://localhost:5000'
    )
    
    # 🔄 Verify the provider meets the contract
    success, logs = verifier.verify_pacts(
        './pacts/orderservice-paymentservice.json'
    )
    
    assert success == 0  # ✅ All contracts satisfied!

💡 Practical Examples

🛒 Example 1: E-Commerce Order API

Let’s build a real contract testing scenario:

# 🛍️ Order service contract testing
import pytest
from pact import Consumer, Provider, Like, EachLike, Term

class TestOrderAPI:
    def setup_method(self):
        # 🎯 Initialize contract
        self.pact = Consumer('WebStore').has_pact_with(
            Provider('OrderAPI'),
            port=8080
        )
        self.pact.start_service()
    
    def teardown_method(self):
        self.pact.stop_service()
    
    def test_create_order(self):
        # 📝 Define the contract
        expected_order = {
            'order_id': Term(r'\d+', '12345'),  # 🔢 Regex matcher
            'status': 'pending',
            'items': EachLike({  # 📦 Array matcher
                'product_id': Like('PROD-001'),
                'quantity': Like(2),
                'price': Like(29.99)
            }),
            'total': Like(59.98),
            'created_at': Term(r'\d{4}-\d{2}-\d{2}', '2024-01-15')
        }
        
        # 🚀 Set up the interaction
        (self.pact
         .given('products exist in inventory')
         .upon_receiving('a request to create order')
         .with_request('POST', '/orders',
                      headers={'Content-Type': 'application/json'},
                      body={
                          'customer_id': 'CUST-123',
                          'items': [
                              {'product_id': 'PROD-001', 'quantity': 2}
                          ]
                      })
         .will_respond_with(201,
                          headers={'Content-Type': 'application/json'},
                          body=expected_order))
        
        # 🎮 Test the interaction
        with self.pact:
            import requests
            response = requests.post(
                f"{self.pact.uri}/orders",
                json={
                    'customer_id': 'CUST-123',
                    'items': [{'product_id': 'PROD-001', 'quantity': 2}]
                },
                headers={'Content-Type': 'application/json'}
            )
            
            # ✅ Verify response
            assert response.status_code == 201
            data = response.json()
            assert data['status'] == 'pending'
            assert len(data['items']) > 0
            print(f"✨ Order created: {data['order_id']}")

🎯 Try it yourself: Add contract tests for order cancellation and status updates!

🎮 Example 2: User Authentication API

Let’s create contracts for authentication:

# 🔐 Authentication contract testing
from datetime import datetime, timedelta
import jwt
from pact import Consumer, Provider, Like, Term

class TestAuthAPI:
    def setup_method(self):
        # 🎯 Set up authentication contract
        self.pact = Consumer('MobileApp').has_pact_with(
            Provider('AuthService'),
            port=9000
        )
        self.pact.start_service()
    
    def teardown_method(self):
        self.pact.stop_service()
    
    def test_user_login(self):
        # 🔑 Expected token format
        expected_response = {
            'access_token': Term(r'^[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+\.[A-Za-z0-9\-_]+$',
                               'eyJ0eXAiOiJKV1QiLCJhbGc...'),  # 🎫 JWT pattern
            'token_type': 'Bearer',
            'expires_in': Like(3600),
            'user': {
                'id': Like('user-123'),
                'email': Like('[email protected]'),
                'roles': EachLike('user')  # 👥 Array of roles
            }
        }
        
        # 📝 Define login interaction
        (self.pact
         .given('user exists with valid credentials')
         .upon_receiving('a login request')
         .with_request('POST', '/auth/login',
                      headers={'Content-Type': 'application/json'},
                      body={
                          'email': '[email protected]',
                          'password': 'SecurePass123!'
                      })
         .will_respond_with(200,
                          headers={'Content-Type': 'application/json'},
                          body=expected_response))
        
        # 🚀 Test the login flow
        with self.pact:
            import requests
            
            # 🔐 Attempt login
            response = requests.post(
                f"{self.pact.uri}/auth/login",
                json={
                    'email': '[email protected]',
                    'password': 'SecurePass123!'
                }
            )
            
            # ✅ Verify successful login
            assert response.status_code == 200
            data = response.json()
            assert 'access_token' in data
            assert data['token_type'] == 'Bearer'
            print("🎉 Login successful!")
    
    def test_token_refresh(self):
        # 🔄 Test token refresh contract
        (self.pact
         .given('a valid refresh token exists')
         .upon_receiving('a token refresh request')
         .with_request('POST', '/auth/refresh',
                      headers={
                          'Content-Type': 'application/json',
                          'Authorization': 'Bearer old.jwt.token'
                      })
         .will_respond_with(200,
                          body={
                              'access_token': Like('new.jwt.token'),
                              'expires_in': Like(3600)
                          }))
        
        with self.pact:
            import requests
            response = requests.post(
                f"{self.pact.uri}/auth/refresh",
                headers={'Authorization': 'Bearer old.jwt.token'}
            )
            
            assert response.status_code == 200
            assert 'access_token' in response.json()
            print("✨ Token refreshed successfully!")

🚀 Advanced Concepts

🧙‍♂️ Advanced Topic 1: State Management

When you’re ready to level up, implement provider states:

# 🎯 Advanced provider state handling
from flask import Flask, jsonify, request
from pact import Provider

app = Flask(__name__)

# 🗃️ In-memory state
users = {}
orders = {}

# 🎨 Provider state setup
@app.route('/_pact/provider_states', methods=['POST'])
def provider_states():
    state = request.json['state']
    
    if state == 'user exists with ID 123':
        # 👤 Set up user state
        users['123'] = {
            'id': '123',
            'name': 'Test User',
            'email': '[email protected]'
        }
        
    elif state == 'order exists with items':
        # 📦 Set up order state
        orders['ORDER-001'] = {
            'id': 'ORDER-001',
            'user_id': '123',
            'items': [
                {'product': 'Widget', 'quantity': 2}
            ],
            'status': 'pending'
        }
    
    return jsonify({'result': state})

# 🚀 API endpoints that use state
@app.route('/users/<user_id>')
def get_user(user_id):
    if user_id in users:
        return jsonify(users[user_id])
    return jsonify({'error': 'User not found'}), 404

# 🔧 Verify contracts with state
def verify_pact_with_states():
    from pact import Verifier
    
    verifier = Verifier(
        provider='UserService',
        provider_base_url='http://localhost:5000'
    )
    
    # 🎯 Point to provider states endpoint
    verifier.provider_states_setup_url = 'http://localhost:5000/_pact/provider_states'
    
    success, logs = verifier.verify_pacts(
        './pacts/consumer-userservice.json'
    )
    
    return success == 0

🏗️ Advanced Topic 2: Message Queue Contracts

For the brave developers working with async messaging:

# 🚀 Message queue contract testing
from pact import MessageConsumer, MessageProvider, Like

class TestMessageContracts:
    def test_order_placed_message(self):
        # 📨 Define message contract
        pact = MessageConsumer('NotificationService').has_pact_with(
            MessageProvider('OrderService'),
            publish_verification_results=True
        )
        
        # 🎯 Expected message format
        expected_message = {
            'event_type': 'order.placed',
            'timestamp': Term(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z',
                            '2024-01-15T10:30:00Z'),
            'data': {
                'order_id': Like('ORDER-123'),
                'customer_email': Like('[email protected]'),
                'total_amount': Like(99.99),
                'items': EachLike({
                    'name': Like('Product'),
                    'quantity': Like(1),
                    'price': Like(49.99)
                })
            }
        }
        
        # 📝 Set up message expectation
        (pact
         .given('an order has been placed')
         .expects_to_receive('an order placed event')
         .with_content(expected_message)
         .with_metadata({'routing_key': 'orders.placed'}))
        
        # 🚀 Handler that processes the message
        def handle_order_placed(message):
            # 📧 Send notification email
            email_data = message['data']
            print(f"📧 Sending order confirmation to {email_data['customer_email']}")
            return True
        
        # ✅ Verify message handling
        with pact:
            # Simulate receiving the message
            result = handle_order_placed(expected_message)
            assert result is True

⚠️ Common Pitfalls and Solutions

😱 Pitfall 1: Over-Specifying Contracts

# ❌ Wrong way - too rigid!
pact.will_respond_with(200, body={
    'user_id': '12345',  # 💥 Hardcoded ID will break!
    'created_at': '2024-01-15 10:30:45',  # 💥 Exact timestamp!
    'session_id': 'abc123def456'  # 💥 Specific session!
})

# ✅ Correct way - flexible matchers!
from pact import Like, Term

pact.will_respond_with(200, body={
    'user_id': Like('12345'),  # 🎯 Any string ID
    'created_at': Term(r'\d{4}-\d{2}-\d{2}', '2024-01-15'),  # 📅 Date pattern
    'session_id': Term(r'[a-f0-9]{12}', 'abc123def456')  # 🔑 Session pattern
})

🤯 Pitfall 2: Ignoring Provider States

# ❌ Dangerous - no state setup!
def test_delete_user():
    pact.upon_receiving('delete user request')
    pact.with_request('DELETE', '/users/999')  # 💥 User might not exist!
    pact.will_respond_with(204)

# ✅ Safe - proper state management!
def test_delete_user():
    pact.given('user exists with ID 999')  # 🎯 Ensure user exists
    pact.upon_receiving('delete user request')
    pact.with_request('DELETE', '/users/999')
    pact.will_respond_with(204)  # ✅ Safe deletion!

🛠️ Best Practices

🎯 Use Matchers Wisely: Don’t over-specify - use Like() and Term() for flexibility
📝 Document Provider States: Clear state names help providers implement correctly
🛡️ Version Your Contracts: Use pact broker for contract versioning
🎨 Keep Contracts Focused: One contract per consumer-provider pair
✨ Test Happy and Sad Paths: Include error scenarios in contracts

🧪 Hands-On Exercise

🎯 Challenge: Build a Weather API Contract Test

Create contract tests for a weather service API:

📋 Requirements:

✅ Get current weather by city name
🏷️ Support multiple temperature units (Celsius/Fahrenheit)
👤 Handle city not found errors
📅 Include forecast endpoint with 5-day data
🎨 Each response needs weather emoji based on conditions!

🚀 Bonus Points:

Add provider state for different weather conditions
Implement contract versioning
Create both consumer and provider tests

💡 Solution

🔍 Click to see solution

# 🎯 Weather API contract testing solution!
from pact import Consumer, Provider, Like, EachLike, Term
import pytest
import requests

class TestWeatherAPIContract:
    def setup_method(self):
        # 🌤️ Set up weather service contract
        self.pact = Consumer('WeatherApp').has_pact_with(
            Provider('WeatherService'),
            port=8888
        )
        self.pact.start_service()
    
    def teardown_method(self):
        self.pact.stop_service()
    
    def test_get_current_weather(self):
        # ☀️ Expected weather response
        expected_weather = {
            'city': Like('London'),
            'temperature': Like(20.5),
            'unit': Term(r'celsius|fahrenheit', 'celsius'),
            'condition': Like('sunny'),
            'emoji': Term(r'[☀️🌤️⛅🌦️🌧️⛈️]', '☀️'),
            'humidity': Like(65),
            'wind_speed': Like(15.5),
            'timestamp': Term(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z',
                            '2024-01-15T12:00:00Z')
        }
        
        # 📝 Define the interaction
        (self.pact
         .given('weather data exists for London')
         .upon_receiving('a request for current weather')
         .with_request('GET', '/weather/current',
                      query={'city': 'London', 'units': 'celsius'})
         .will_respond_with(200,
                          headers={'Content-Type': 'application/json'},
                          body=expected_weather))
        
        # 🚀 Test the interaction
        with self.pact:
            response = requests.get(
                f"{self.pact.uri}/weather/current",
                params={'city': 'London', 'units': 'celsius'}
            )
            
            assert response.status_code == 200
            data = response.json()
            assert data['city'] == 'London'
            assert 'emoji' in data
            print(f"🌡️ Current weather: {data['temperature']}°C {data['emoji']}")
    
    def test_get_weather_forecast(self):
        # 📅 5-day forecast contract
        expected_forecast = {
            'city': Like('Paris'),
            'forecast': EachLike({
                'date': Term(r'\d{4}-\d{2}-\d{2}', '2024-01-16'),
                'high': Like(25.0),
                'low': Like(15.0),
                'condition': Like('partly cloudy'),
                'emoji': Like('⛅'),
                'precipitation_chance': Like(30)
            }, minimum=5)
        }
        
        (self.pact
         .given('forecast data exists for Paris')
         .upon_receiving('a request for 5-day forecast')
         .with_request('GET', '/weather/forecast',
                      query={'city': 'Paris', 'days': '5'})
         .will_respond_with(200, body=expected_forecast))
        
        with self.pact:
            response = requests.get(
                f"{self.pact.uri}/weather/forecast",
                params={'city': 'Paris', 'days': '5'}
            )
            
            assert response.status_code == 200
            data = response.json()
            assert len(data['forecast']) >= 5
            print("📊 5-day forecast received!")
    
    def test_city_not_found(self):
        # ❌ Error scenario contract
        error_response = {
            'error': 'city_not_found',
            'message': Like('City not found: InvalidCity'),
            'suggestions': EachLike('London')
        }
        
        (self.pact
         .given('city does not exist in database')
         .upon_receiving('a request for invalid city')
         .with_request('GET', '/weather/current',
                      query={'city': 'InvalidCity'})
         .will_respond_with(404, body=error_response))
        
        with self.pact:
            response = requests.get(
                f"{self.pact.uri}/weather/current",
                params={'city': 'InvalidCity'}
            )
            
            assert response.status_code == 404
            data = response.json()
            assert data['error'] == 'city_not_found'
            print("⚠️ City not found handled correctly!")

# 🎮 Provider verification
def verify_weather_provider():
    from pact import Verifier
    
    verifier = Verifier(
        provider='WeatherService',
        provider_base_url='http://localhost:5000'
    )
    
    # 🎯 Add provider states
    verifier.provider_states_setup_url = 'http://localhost:5000/_pact/provider_states'
    
    success, logs = verifier.verify_pacts(
        './pacts/weatherapp-weatherservice.json',
        verbose=True
    )
    
    print("✅ All weather contracts verified!" if success == 0 else "❌ Contract verification failed!")
    return success == 0

🎓 Key Takeaways

You’ve learned so much! Here’s what you can now do:

✅ Create contract tests between services with confidence 💪
✅ Use flexible matchers to avoid brittle tests 🛡️
✅ Implement provider states for realistic testing 🎯
✅ Debug contract failures like a pro 🐛
✅ Build reliable microservices with contract testing! 🚀

Remember: Contract testing is your safety net for distributed systems. It catches integration issues before they become production nightmares! 🤝

🤝 Next Steps

Congratulations! 🎉 You’ve mastered contract testing for API contracts!

Here’s what to do next:

💻 Practice with the weather API exercise above
🏗️ Add contract tests to your existing APIs
📚 Move on to our next tutorial: Property-Based Testing
🌟 Set up a Pact Broker for team collaboration!

Remember: Every microservice expert started with their first contract test. Keep testing, keep learning, and most importantly, keep your APIs reliable! 🚀

Happy contract testing! 🎉🚀✨

Prerequisites

What you'll learn