Simple Contract Testing

---

This guide shows how to test API contracts by validating response schemas and status codes. Perfect for ensuring API consistency and catching breaking changes.

Use Cases

• Validate API response schemas
• Check expected status codes
• Test required response fields
• Ensure API contract compliance

Simple Implementation

from locust import task, HttpUser
import json

class ContractTestUser(HttpUser):
    def on_start(self):
        # Expected API contracts
        self.contracts = {
            "/api/users": {
                "status_code": 200,
                "required_fields": ["id", "name", "email"],
                "field_types": {
                    "id": int,
                    "name": str,
                    "email": str
                }
            },
            "/api/products": {
                "status_code": 200,
                "required_fields": ["id", "name", "price"],
                "field_types": {
                    "id": int,
                    "name": str,
                    "price": (int, float)
                }
            },
            "/api/orders": {
                "status_code": 200,
                "required_fields": ["id", "user_id", "total"],
                "field_types": {
                    "id": int,
                    "user_id": int,
                    "total": (int, float)
                }
            }
        }

    @task(3)
    def test_users_contract(self):
        """Test users API contract"""
        endpoint = "/api/users"
        
        with self.client.get(endpoint, name="Users Contract") as response:
            self.validate_contract(endpoint, response)

    @task(2)
    def test_products_contract(self):
        """Test products API contract"""
        endpoint = "/api/products"
        
        with self.client.get(endpoint, name="Products Contract") as response:
            self.validate_contract(endpoint, response)

    @task(2)
    def test_orders_contract(self):
        """Test orders API contract"""
        endpoint = "/api/orders"
        
        with self.client.get(endpoint, name="Orders Contract") as response:
            self.validate_contract(endpoint, response)

    @task(1)
    def test_single_user_contract(self):
        """Test single user API contract"""
        endpoint = "/api/users/1"
        
        with self.client.get(endpoint, name="Single User Contract") as response:
            # Use same contract as users list but for single item
            if response.status_code == 200:
                try:
                    data = response.json()
                    self.validate_single_item_schema(data, self.contracts["/api/users"])
                    print("Single user contract: PASSED")
                except Exception as e:
                    response.failure(f"Single user contract failed: {e}")
            elif response.status_code == 404:
                print("Single user contract: User not found (acceptable)")
            else:
                response.failure(f"Unexpected status code: {response.status_code}")

    def validate_contract(self, endpoint, response):
        """Validate API response against contract"""
        if endpoint not in self.contracts:
            response.failure(f"No contract defined for {endpoint}")
            return
            
        contract = self.contracts[endpoint]
        
        # Check status code
        if response.status_code != contract["status_code"]:
            response.failure(f"Status code mismatch: expected {contract['status_code']}, got {response.status_code}")
            return
        
        # Parse JSON response
        try:
            data = response.json()
        except json.JSONDecodeError:
            response.failure("Invalid JSON response")
            return
        
        # Handle both single items and arrays
        if isinstance(data, list):
            if len(data) > 0:
                self.validate_single_item_schema(data[0], contract)
            print(f"Contract validation for {endpoint}: PASSED (array with {len(data)} items)")
        else:
            self.validate_single_item_schema(data, contract)
            print(f"Contract validation for {endpoint}: PASSED")

    def validate_single_item_schema(self, item, contract):
        """Validate a single item against the contract schema"""
        # Check required fields
        for field in contract["required_fields"]:
            if field not in item:
                raise ValueError(f"Missing required field: {field}")
        
        # Check field types
        for field, expected_type in contract["field_types"].items():
            if field in item:
                if not isinstance(item[field], expected_type):
                    raise ValueError(f"Field {field} has wrong type: expected {expected_type}, got {type(item[field])}")

    @task(1)
    def test_error_responses(self):
        """Test error response contracts"""
        # Test 404 response
        with self.client.get("/api/users/99999", name="404 Error Contract") as response:
            if response.status_code == 404:
                try:
                    error_data = response.json()
                    # Basic error response structure
                    if "error" in error_data or "message" in error_data:
                        print("404 error contract: PASSED")
                    else:
                        response.failure("404 response missing error message")
                except json.JSONDecodeError:
                    response.failure("404 response not valid JSON")
            else:
                response.failure(f"Expected 404, got {response.status_code}")

    @task(1)
    def test_post_contract(self):
        """Test POST request contract"""
        user_data = {
            "name": "Test User",
            "email": "test@example.com"
        }
        
        with self.client.post(
            "/api/users",
            json=user_data,
            name="POST Contract"
        ) as response:
            if response.status_code in [201, 200]:
                try:
                    created_user = response.json()
                    
                    # Validate created user has required fields
                    required_fields = ["id", "name", "email"]
                    for field in required_fields:
                        if field not in created_user:
                            response.failure(f"Created user missing field: {field}")
                            return
                    
                    # Validate the data we sent is reflected
                    if created_user["name"] != user_data["name"]:
                        response.failure("Created user name doesn't match input")
                        return
                    
                    if created_user["email"] != user_data["email"]:
                        response.failure("Created user email doesn't match input")
                        return
                    
                    print("POST contract validation: PASSED")
                    
                except json.JSONDecodeError:
                    response.failure("POST response not valid JSON")
            elif response.status_code == 400:
                # Bad request is acceptable for validation errors
                print("POST contract: Validation error (acceptable)")
            else:
                response.failure(f"Unexpected POST status code: {response.status_code}")

Setup Instructions

1. Define your API contracts with expected fields and types
2. Add endpoints you want to test to the contracts dictionary
3. Customize field validation rules for your specific API
4. Run tests to catch contract violations

What This Tests

• Response Schemas: Validates JSON structure and required fields
• Status Codes: Ensures endpoints return expected HTTP codes
• Field Types: Checks that fields have correct data types
• Error Handling: Tests error response formats

Contract Validation

The guide validates:

• Required Fields: Ensures critical fields are always present
• Data Types: Validates field types (string, integer, etc.)
• Status Codes: Checks HTTP response codes match expectations
• Error Responses: Validates error message structure

Common Contract Issues

• Missing Fields: Required fields not in response
• Wrong Types: Fields with unexpected data types
• Status Code Changes: Endpoints returning different codes
• Schema Drift: API responses changing over time

Extending Contracts

Add more validation rules:

• Field Formats: Email, date, URL validation
• Value Ranges: Min/max values for numbers
• Enum Values: Restricted field values
• Nested Objects: Complex object validation