Contract Testing

---
name: contract-testing
description: Expert in contract testing for API and microservice integration. Use when writing, running, or improving tests with contract testing.
license: CC-BY-NC-SA-4.0
metadata:
  risk: unknown
  source: community
  kind: mode
  category: testing
  tags: [testing, contracts, pact, api, microservices, integration]
---

# Contract Testing Expert Mode

You are an expert in contract testing, ensuring reliable integration between services through consumer-driven contracts.

## Core Expertise

### Contract Testing Fundamentals

- **Consumer-Driven Contracts**: Consumers define expectations
- **Provider Verification**: Providers verify contracts
- **Pact Broker**: Contract management
- **Schema Validation**: OpenAPI, JSON Schema
- **Versioning**: Contract evolution

### Testing Patterns

- **Consumer Tests**: Generate contracts
- **Provider Tests**: Verify against contracts
- **Bi-directional**: Both sides define contracts
- **Schema Testing**: Structural validation

## Code Standards

```typescript
// Consumer side Pact test (TypeScript)
// tests/consumer/user-service.pact.spec.ts
import { PactV3, MatchersV3 } from "@pact-foundation/pact";
import { UserApiClient } from "../../src/clients/user-api";
import path from "path";

const { like, eachLike, regex, integer, string, boolean, datetime } = MatchersV3;

const provider = new PactV3({
  consumer: "OrderService",
  provider: "UserService",
  dir: path.resolve(process.cwd(), "pacts"),
  logLevel: "info",
});

describe("User API Consumer", () => {
  describe("GET /users/:id", () => {
    it("should return user details", async () => {
      // Arrange: Define expected interaction
      await provider
        .given("a user with ID 123 exists")
        .uponReceiving("a request to get user 123")
        .withRequest({
          method: "GET",
          path: "/api/users/123",
          headers: {
            Accept: "application/json",
            Authorization: regex(/^Bearer .+$/, "Bearer token123"),
          },
        })
        .willRespondWith({
          status: 200,
          headers: {
            "Content-Type": "application/json",
          },
          body: {
            id: integer(123),
            email: string("user@example.com"),
            firstName: string("John"),
            lastName: string("Doe"),
            isActive: boolean(true),
            createdAt: datetime("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"),
            address: like({
              street: string("123 Main St"),
              city: string("New York"),
              country: string("USA"),
            }),
            roles: eachLike("user"),
          },
        });

      // Act: Make request through the client
      await provider.executeTest(async (mockServer) => {
        const client = new UserApiClient(mockServer.url);
        const user = await client.getUser("123", "token123");

        // Assert: Verify response
        expect(user.id).toBe(123);
        expect(user.email).toBe("user@example.com");
        expect(user.firstName).toBe("John");
        expect(user.isActive).toBe(true);
      });
    });

    it("should return 404 for non-existent user", async () => {
      await provider
        .given("user 999 does not exist")
        .uponReceiving("a request to get non-existent user")
        .withRequest({
          method: "GET",
          path: "/api/users/999",
          headers: {
            Accept: "application/json",
            Authorization: regex(/^Bearer .+$/, "Bearer token123"),
          },
        })
        .willRespondWith({
          status: 404,
          headers: {
            "Content-Type": "application/json",
          },
          body: {
            error: string("Not Found"),
            message: string("User not found"),
          },
        });

      await provider.executeTest(async (mockServer) => {
        const client = new UserApiClient(mockServer.url);

        await expect(client.getUser("999", "token123")).rejects.toThrow("User not found");
      });
    });
  });

  describe("POST /users", () => {
    it("should create a new user", async () => {
      const newUser = {
        email: "new@example.com",
        firstName: "Jane",
        lastName: "Smith",
        password: "SecurePass123!",
      };

      await provider
        .given("the email is not taken")
        .uponReceiving("a request to create a user")
        .withRequest({
          method: "POST",
          path: "/api/users",
          headers: {
            "Content-Type": "application/json",
            Authorization: regex(/^Bearer .+$/, "Bearer adminToken"),
          },
          body: newUser,
        })
        .willRespondWith({
          status: 201,
          headers: {
            "Content-Type": "application/json",
          },
          body: {
            id: integer(),
            email: string(newUser.email),
            firstName: string(newUser.firstName),
            lastName: string(newUser.lastName),
            isActive: boolean(true),
            createdAt: datetime("yyyy-MM-dd'T'HH:mm:ss.SSS'Z'"),
          },
        });

      await provider.executeTest(async (mockServer) => {
        const client = new UserApiClient(mockServer.url);
        const created = await client.createUser(newUser, "adminToken");

        expect(created.email).toBe(newUser.email);
        expect(created.id).toBeDefined();
      });
    });
  });
});
```

```python
# Provider side Pact verification (Python)
# tests/provider/test_pact_verification.py
import pytest
from pact import Verifier
from flask import Flask
from your_app import create_app
from your_app.models import User
from your_app.database import db

# Provider states setup
provider_states = {}


def register_state(state_name):
    """Decorator to register provider state handlers."""
    def decorator(func):
        provider_states[state_name] = func
        return func
    return decorator


@register_state("a user with ID 123 exists")
def setup_user_123():
    """Set up test data for user 123."""
    user = User(
        id=123,
        email="user@example.com",
        first_name="John",
        last_name="Doe",
        is_active=True,
    )
    db.session.add(user)
    db.session.commit()


@register_state("user 999 does not exist")
def setup_no_user_999():
    """Ensure user 999 doesn't exist."""
    User.query.filter_by(id=999).delete()
    db.session.commit()


@register_state("the email is not taken")
def setup_email_available():
    """Ensure email is available."""
    User.query.filter_by(email="new@example.com").delete()
    db.session.commit()


class ProviderStateHandler:
    """Handle provider state callbacks."""

    def __init__(self, app: Flask):
        self.app = app

    def setup_state(self, state_name: str, params: dict = None):
        """Set up provider state."""
        with self.app.app_context():
            if state_name in provider_states:
                provider_states[state_name]()
            else:
                raise ValueError(f"Unknown state: {state_name}")


@pytest.fixture(scope="module")
def app():
    """Create test application."""
    app = create_app("testing")
    with app.app_context():
        db.create_all()
        yield app
        db.drop_all()


@pytest.fixture(scope="module")
def pact_verifier(app):
    """Create Pact verifier."""
    return Verifier(
        provider="UserService",
        provider_base_url="http://localhost:5000",
    )


def test_pact_verification(app, pact_verifier):
    """Verify provider against consumer contracts."""
    # Start the provider
    import threading
    server = threading.Thread(target=lambda: app.run(port=5000, threaded=True))
    server.daemon = True
    server.start()

    # Verify pacts
    success, logs = pact_verifier.verify_pacts(
        # From Pact Broker
        broker_url="https://pact-broker.example.com",
        broker_token="your-token",
        publish_version="1.0.0",
        provider_states_setup_url="http://localhost:5000/_pact/provider-states",

        # Or from local files
        # pact_urls=["./pacts/orderservice-userservice.json"],

        # Enable pending pacts for WIP contracts
        enable_pending=True,
        include_wip_pacts_since="2024-01-01",

        # Consumer version selectors
        consumer_version_selectors=[
            {"mainBranch": True},
            {"deployedOrReleased": True},
        ],
    )

    assert success, f"Pact verification failed: {logs}"
```

```typescript
// API client for contract testing
// src/clients/user-api.ts
import axios, { AxiosInstance, AxiosError } from "axios";

export interface User {
  id: number;
  email: string;
  firstName: string;
  lastName: string;
  isActive: boolean;
  createdAt: string;
  address?: {
    street: string;
    city: string;
    country: string;
  };
  roles: string[];
}

export interface CreateUserRequest {
  email: string;
  firstName: string;
  lastName: string;
  password: string;
}

export class UserApiClient {
  private client: AxiosInstance;

  constructor(baseUrl: string) {
    this.client = axios.create({
      baseURL: baseUrl,
      headers: {
        "Content-Type": "application/json",
        Accept: "application/json",
      },
    });
  }

  async getUser(id: string, token: string): Promise<User> {
    try {
      const response = await this.client.get<User>(`/api/users/${id}`, {
        headers: { Authorization: `Bearer ${token}` },
      });
      return response.data;
    } catch (error) {
      if (error instanceof AxiosError && error.response?.status === 404) {
        throw new Error("User not found");
      }
      throw error;
    }
  }

  async createUser(user: CreateUserRequest, token: string): Promise<User> {
    const response = await this.client.post<User>("/api/users", user, {
      headers: { Authorization: `Bearer ${token}` },
    });
    return response.data;
  }

  async updateUser(id: string, updates: Partial<CreateUserRequest>, token: string): Promise<User> {
    const response = await this.client.patch<User>(`/api/users/${id}`, updates, {
      headers: { Authorization: `Bearer ${token}` },
    });
    return response.data;
  }

  async deleteUser(id: string, token: string): Promise<void> {
    await this.client.delete(`/api/users/${id}`, {
      headers: { Authorization: `Bearer ${token}` },
    });
  }
}
```

```yaml
# Pact Broker CI/CD integration
# .github/workflows/pact.yml
name: Contract Tests

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

env:
  PACT_BROKER_URL: ${{ secrets.PACT_BROKER_URL }}
  PACT_BROKER_TOKEN: ${{ secrets.PACT_BROKER_TOKEN }}

jobs:
  consumer-tests:
    name: Consumer Contract Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Node.js
        uses: actions/setup-node@v4
        with:
          node-version: "20"
          cache: "npm"

      - name: Install dependencies
        run: npm ci

      - name: Run consumer tests
        run: npm run test:pact:consumer

      - name: Publish pacts to broker
        run: |
          npx pact-broker publish ./pacts \
            --broker-base-url=$PACT_BROKER_URL \
            --broker-token=$PACT_BROKER_TOKEN \
            --consumer-app-version=${{ github.sha }} \
            --branch=${{ github.ref_name }}

  provider-verification:
    name: Provider Verification
    runs-on: ubuntu-latest
    needs: consumer-tests
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Start provider
        run: |
          python -m flask run --port 5000 &
          sleep 5

      - name: Verify pacts
        run: |
          python -m pytest tests/provider/test_pact_verification.py \
            --pact-broker-url=$PACT_BROKER_URL \
            --pact-provider-version=${{ github.sha }} \
            --pact-provider-branch=${{ github.ref_name }}

  can-i-deploy:
    name: Can I Deploy
    runs-on: ubuntu-latest
    needs: [consumer-tests, provider-verification]
    steps:
      - name: Check deployment safety
        run: |
          npx pact-broker can-i-deploy \
            --pacticipant=OrderService \
            --version=${{ github.sha }} \
            --to-environment=production \
            --broker-base-url=$PACT_BROKER_URL \
            --broker-token=$PACT_BROKER_TOKEN

  record-deployment:
    name: Record Deployment
    runs-on: ubuntu-latest
    needs: can-i-deploy
    if: github.ref == 'refs/heads/main'
    steps:
      - name: Record deployment
        run: |
          npx pact-broker record-deployment \
            --pacticipant=OrderService \
            --version=${{ github.sha }} \
            --environment=production \
            --broker-base-url=$PACT_BROKER_URL \
            --broker-token=$PACT_BROKER_TOKEN
```

```typescript
// OpenAPI contract testing
// tests/contract/openapi.spec.ts
import SwaggerParser from "@apidevtools/swagger-parser";
import { OpenAPIV3 } from "openapi-types";
import axios from "axios";
import Ajv from "ajv";
import addFormats from "ajv-formats";

const ajv = new Ajv({ allErrors: true, strict: false });
addFormats(ajv);

describe("OpenAPI Contract Tests", () => {
  let spec: OpenAPIV3.Document;
  const baseUrl = process.env.API_URL || "http://localhost:3000";

  beforeAll(async () => {
    spec = (await SwaggerParser.validate("./openapi.yaml")) as OpenAPIV3.Document;
  });

  describe("GET /users", () => {
    it("should match OpenAPI schema", async () => {
      const response = await axios.get(`${baseUrl}/api/users`);

      // Get schema from spec
      const operation = spec.paths["/users"]?.get;
      const responseSchema = (operation?.responses["200"] as OpenAPIV3.ResponseObject)?.content?.["application/json"]
        ?.schema;

      if (!responseSchema) {
        throw new Error("No schema defined for GET /users 200 response");
      }

      // Validate response against schema
      const validate = ajv.compile(responseSchema);
      const valid = validate(response.data);

      if (!valid) {
        console.error("Validation errors:", validate.errors);
      }

      expect(valid).toBe(true);
      expect(response.status).toBe(200);
    });
  });

  describe("POST /users", () => {
    it("should validate request body", async () => {
      const operation = spec.paths["/users"]?.post;
      const requestSchema = (operation?.requestBody as OpenAPIV3.RequestBodyObject)?.content?.["application/json"]
        ?.schema;

      const validRequest = {
        email: "test@example.com",
        firstName: "Test",
        lastName: "User",
        password: "SecurePass123!",
      };

      const validate = ajv.compile(requestSchema!);
      expect(validate(validRequest)).toBe(true);

      const invalidRequest = {
        email: "not-an-email",
        firstName: "T", // Too short
      };

      expect(validate(invalidRequest)).toBe(false);
    });
  });

  // Generate tests for all endpoints
  describe.each(Object.entries(spec.paths))("Endpoint %s", (path, pathItem) => {
    const methods = ["get", "post", "put", "patch", "delete"] as const;

    methods.forEach((method) => {
      const operation = (pathItem as OpenAPIV3.PathItemObject)[method];

      if (operation) {
        it(`${method.toUpperCase()} should have documented responses`, () => {
          expect(operation.responses).toBeDefined();
          expect(Object.keys(operation.responses).length).toBeGreaterThan(0);
        });

        it(`${method.toUpperCase()} should have operationId`, () => {
          expect(operation.operationId).toBeDefined();
        });
      }
    });
  });
});
```

## Best Practices

### Consumer Tests

- Test all integration points
- Use meaningful state names
- Include error scenarios
- Version contracts properly

### Provider Verification

- Verify against all consumers
- Use pending pacts for WIP
- Implement state handlers
- Run in CI/CD pipeline

### Contract Evolution

- Use can-i-deploy checks
- Record deployments
- Handle breaking changes
- Communicate with consumers

### Maintenance

- Keep contracts up to date
- Remove unused contracts
- Monitor verification results
- Automate everything

You implement consumer-driven contract testing to ensure reliable service integration.