Integration Test Documentation: Comprehensive Guide to API Contracts and System Interfaces

# user-service-contract.yaml
openapi: 3.0.0
info:
  title: User Service API
  version: 2.1.0
  description: User management and authentication service

paths:
  /api/users/{userId}:
    get:
      summary: Retrieve user details
      operationId: getUserById
      parameters:
        - name: userId
          in: path
          required: true
          schema:
            type: string
            format: uuid
      responses:
        '200':
          description: User found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/User'
              example:
                id: "550e8400-e29b-41d4-a716-446655440000"
                email: "user@example.com"
                firstName: "John"
                lastName: "Doe"
                status: "active"
                createdAt: "2024-01-15T10:30:00Z"
        '404':
          description: User not found
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/Error'
        '500':
          description: Internal server error

  /api/users:
    post:
      summary: Create new user
      requestBody:
        required: true
        content:
          application/json:
            schema:
              $ref: '#/components/schemas/CreateUserRequest'
      responses:
        '201':
          description: User created
          content:
            application/json:
              schema:
                $ref: '#/components/schemas/User'
        '400':
          description: Invalid request
        '409':
          description: User already exists

components:
  schemas:
    User:
      type: object
      required:
        - id
        - email
        - status
      properties:
        id:
          type: string
          format: uuid
        email:
          type: string
          format: email
        firstName:
          type: string
        lastName:
          type: string
        status:
          type: string
          enum: [active, inactive, suspended]
        createdAt:
          type: string
          format: date-time

# test_user_service_integration.py
import pytest
import requests
from jsonschema import validate
from datetime import datetime

class TestUserServiceIntegration:
    """
    Integration tests for User Service API
    Based on contract: user-service-contract.yaml v2.1.0
    """

    BASE_URL = "https://api-staging.example.com"

    @pytest.fixture
    def auth_headers(self):
        """Get authentication headers for tests"""
        return {
            "Authorization": "Bearer test_token",
            "Content-Type": "application/json"
        }

    def test_get_user_success_contract(self, auth_headers):
        """
        Verify GET /api/users/{userId} returns data matching contract
        Contract: 200 response must match User schema
        """
        user_id = "550e8400-e29b-41d4-a716-446655440000"
        response = requests.get(
            f"{self.BASE_URL}/api/users/{user_id}",
            headers=auth_headers
        )

        # Verify status code per contract
        assert response.status_code == 200

        # Verify response structure matches contract
        data = response.json()
        assert "id" in data
        assert "email" in data
        assert "status" in data

        # Verify data types
        assert isinstance(data["id"], str)
        assert isinstance(data["email"], str)
        assert data["status"] in ["active", "inactive", "suspended"]

        # Verify UUID format
        from uuid import UUID
        UUID(data["id"])  # Raises ValueError if invalid

    def test_get_user_not_found_contract(self, auth_headers):
        """
        Verify GET /api/users/{userId} returns 404 for non-existent user
        Contract: 404 response with Error schema
        """
        non_existent_id = "00000000-0000-0000-0000-000000000000"
        response = requests.get(
            f"{self.BASE_URL}/api/users/{non_existent_id}",
            headers=auth_headers
        )

        assert response.status_code == 404
        assert "error" in response.json() or "message" in response.json()

    def test_create_user_contract_validation(self, auth_headers):
        """
        Verify POST /api/users validates request per contract
        Contract: 400 for invalid request, 201 for success
        """
        # Test invalid email format
        invalid_payload = {
            "email": "not-an-email",
            "firstName": "John",
            "lastName": "Doe"
        }
        response = requests.post(
            f"{self.BASE_URL}/api/users",
            json=invalid_payload,
            headers=auth_headers
        )
        assert response.status_code == 400

        # Test valid creation
        valid_payload = {
            "email": f"test_{datetime.now().timestamp()}@example.com",
            "firstName": "John",
            "lastName": "Doe"
        }
        response = requests.post(
            f"{self.BASE_URL}/api/users",
            json=valid_payload,
            headers=auth_headers
        )
        assert response.status_code == 201

        # Verify created user matches contract
        created_user = response.json()
        assert created_user["email"] == valid_payload["email"]
        assert created_user["status"] == "active"  # Default per contract

# System Integration Map

## Architecture Overview

# service-dependencies.yaml
services:
  api-gateway:
    depends_on:
      - user-service
      - order-service
      - product-service
    integration_tests:
      - test_user_authentication_flow
      - test_order_placement_flow
      - test_product_search_flow

  order-service:
    depends_on:
      - user-service (authentication)
      - inventory-service (stock validation)
      - payment-gateway (external)
      - notification-service (order confirmation)
    integration_tests:
      - test_order_creation_with_auth
      - test_inventory_validation
      - test_payment_processing
      - test_order_notifications
    failure_scenarios:
      - inventory_service_down: "Fallback to cached stock levels"
      - payment_gateway_timeout: "Queue for retry, notify user"
      - notification_service_down: "Queue notifications, order proceeds"

  payment-service:
    depends_on:
      - stripe-api (external)
      - paypal-api (external)
    integration_tests:
      - test_stripe_integration
      - test_paypal_integration
      - test_payment_provider_failover
    circuit_breaker:
      enabled: true
      failure_threshold: 5
      timeout: 30s
      half_open_requests: 3

## Order Creation Data Flow

### 1. Request Flow

# test_order_flow_integration.py
import pytest
from unittest.mock import Mock, patch
import json

class TestOrderCreationDataFlow:
    """
    End-to-end integration test for order creation data flow
    Tests data transformation across service boundaries
    """

    def test_complete_order_flow_data_transformation(self):
        """
        Verify data transforms correctly through entire order flow
        """
        # 1. Client request
        client_request = {
            "items": [
                {"productId": "PROD-001", "quantity": 2},
                {"productId": "PROD-002", "quantity": 1}
            ],
            "shipping": {
                "address": "123 Main St",
                "city": "Springfield",
                "zipCode": "12345"
            }
        }

        # 2. API Gateway enrichment
        with patch('auth_service.verify_token') as mock_auth:
            mock_auth.return_value = {
                "userId": "USER-123",
                "email": "user@example.com",
                "name": "John Doe"
            }

            enriched_request = self.api_gateway.enrich_request(
                client_request,
                headers={"Authorization": "Bearer test_token"}
            )

            # Verify enrichment
            assert enriched_request["userId"] == "USER-123"
            assert "timestamp" in enriched_request
            assert "requestId" in enriched_request

        # 3. Inventory check
        with patch('inventory_service.check_stock') as mock_inventory:
            mock_inventory.return_value = {
                "PROD-001": {"available": True, "reservationId": "RES-001"},
                "PROD-002": {"available": True, "reservationId": "RES-002"}
            }

            inventory_response = self.order_service.validate_inventory(
                enriched_request["items"]
            )

            # Verify inventory data structure
            assert all(item["available"] for item in inventory_response.values())

        # 4. Order creation
        order_data = self.order_service.create_order(enriched_request)

        # Verify order data structure
        assert order_data["orderId"]
        assert order_data["status"] == "pending"
        assert order_data["userId"] == "USER-123"
        assert len(order_data["items"]) == 2

        # 5. Payment processing
        with patch('payment_gateway.process_payment') as mock_payment:
            mock_payment.return_value = {
                "transactionId": "TXN-999",
                "status": "authorized",
                "authCode": "AUTH123"
            }

            payment_request = {
                "amount": order_data["total"],
                "currency": "USD",
                "orderId": order_data["orderId"]
            }

            payment_response = self.payment_service.process(payment_request)

            # Verify payment data
            assert payment_response["status"] == "authorized"
            assert payment_response["transactionId"]

        # 6. Order confirmation
        confirmed_order = self.order_service.confirm_order(
            order_data["orderId"],
            payment_response["transactionId"]
        )

        # Verify final order state
        assert confirmed_order["status"] == "confirmed"
        assert confirmed_order["paymentDetails"]["transactionId"] == "TXN-999"

        # 7. Notification data
        notification_payload = self.notification_service.prepare_confirmation(
            confirmed_order
        )

        # Verify notification includes all necessary data
        assert notification_payload["to"] == "user@example.com"
        assert notification_payload["template"] == "order_confirmation"
        assert notification_payload["data"]["orderId"] == order_data["orderId"]
        assert notification_payload["data"]["customerName"] == "John Doe"

# test_circuit_breaker_integration.py
import pytest
from datetime import datetime, timedelta

class TestCircuitBreakerIntegration:
    """
    Test circuit breaker behavior across service boundaries
    """

    def test_payment_service_circuit_breaker(self):
        """
        Verify circuit breaker opens after threshold failures
        and prevents cascading failures
        """
        # Configure circuit breaker
        circuit_breaker_config = {
            "failure_threshold": 5,
            "timeout": 30,  # seconds
            "half_open_max_requests": 3
        }

        # Simulate 5 consecutive failures
        for i in range(5):
            response = self.order_service.process_payment({
                "orderId": f"ORDER-{i}",
                "amount": 100.00,
                "force_failure": True  # Test flag
            })
            assert response["status"] == "failed"

        # 6th request should be rejected by circuit breaker (OPEN state)
        response = self.order_service.process_payment({
            "orderId": "ORDER-6",
            "amount": 100.00
        })
        assert response["status"] == "circuit_open"
        assert "Payment service temporarily unavailable" in response["message"]

        # Verify circuit breaker state
        cb_state = self.order_service.get_circuit_breaker_state("payment_service")
        assert cb_state["state"] == "OPEN"
        assert cb_state["failure_count"] >= 5

    def test_circuit_breaker_half_open_recovery(self):
        """
        Verify circuit breaker transitions to HALF_OPEN and recovers
        """
        # Set circuit to OPEN state
        self._open_circuit_breaker("inventory_service")

        # Wait for timeout period
        time.sleep(31)  # timeout is 30 seconds

        # First request after timeout (HALF_OPEN state)
        response = self.order_service.check_inventory({
            "productId": "PROD-001",
            "quantity": 1
        })

        # Should allow request through
        assert response["status"] == "success"

        # Verify state is HALF_OPEN
        cb_state = self.order_service.get_circuit_breaker_state("inventory_service")
        assert cb_state["state"] == "HALF_OPEN"

        # Send 2 more successful requests (half_open_max_requests = 3)
        for i in range(2):
            response = self.order_service.check_inventory({
                "productId": f"PROD-{i}",
                "quantity": 1
            })
            assert response["status"] == "success"

        # Circuit should now be CLOSED (recovered)
        cb_state = self.order_service.get_circuit_breaker_state("inventory_service")
        assert cb_state["state"] == "CLOSED"
        assert cb_state["failure_count"] == 0

# Integration Test Documentation Template

## 1. Integration Overview
- **Services Involved**: List all systems in integration
- **Integration Type**: Synchronous/Asynchronous, REST/gRPC/Message Queue
- **Business Context**: Why this integration exists
- **Criticality**: P0/P1/P2 classification

## 2. Contract Specification
- API contract (OpenAPI/Protobuf/AsyncAPI)
- Request/Response examples
- Error responses
- Authentication requirements

## 3. Test Scenarios
- Happy path scenarios
- Error scenarios
- Edge cases
- Performance requirements

## 4. Data Requirements
- Test data setup
- Data dependencies
- Cleanup procedures

## 5. Environment Setup
- Service dependencies
- Configuration requirements
- Mock services (if any)

## 6. Execution Instructions
- Pre-requisites
- Execution commands
- Expected results
- Troubleshooting guide