API Mocking for AI Applications: Complete Guide to APIMock-AI Patterns

Build robust AI applications with reliable API mocking — contract testing, chaos engineering, and production-grade mock servers.

Introduction

API mocking is critical for AI applications that depend on external LLM APIs, vector databases, and third-party services. This guide covers production mocking patterns using APIMock-AI principles.

Why API Mocking Matters for AI

WITHOUT MOCKS                    WITH MOCKS
─────────────────────            ─────────────────────
Flaky CI/CD                      Reliable tests
Rate limit failures              Controlled responses
Costly API calls                 Zero-cost development
No offline development           Full offline workflow
Slow feedback loops              Instant feedback
Vendor lock-in risk              Portable test suites

Part 1: Mock Server Architecture

Core Mock Server

# mocks/server.py
from dataclasses import dataclass, field
from typing: Dict, List, Optional, Callable, Any
from enum import Enum
import json
import asyncio
from aiohttp import web
import uuid

class MockStrategy(Enum):
    STATIC = "static"                    # Fixed response
    SEQUENCE = "sequence"                # Rotating responses
    TEMPLATE = "template"                # Jinja2 template
    CALLBACK = "callback"                # Dynamic function
    PROXY = "proxy"                      # Forward to real API
    CHAOS = "chaos"                      # Inject failures

@dataclass
class MockRule:
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    method: str
    path_pattern: str
    strategy: MockStrategy = MockStrategy.STATIC
    response: Any = None
    sequence: List[Any] = field(default_factory=list)
    template: str = ""
    callback: Optional[Callable] = None
    proxy_target: str = ""
    chaos_config: Dict = field(default_factory=dict)
    conditions: Dict = field(default_factory=dict)  # headers, query, body
    priority: int = 0
    times_matched: int = 0
    active: bool = True

class MockServer:
    def __init__(self, host: str = "localhost", port: int = 8080):
        self.host = host
        self.port = port
        self.rules: List[MockRule] = []
        self.request_log: List[Dict] = []
        self.app = web.Application()
        self.app.router.add_route("*", "/{path:.*}", self.handle_request)

    def add_rule(self, rule: MockRule):
        self.rules.append(rule)
        self.rules.sort(key=lambda r: -r.priority)

    def add_static(self, method: str, path: str, response: Any, status: int = 200, headers: Dict = None):
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.STATIC,
            response={"status": status, "body": response, "headers": headers or {}}
        )
        self.add_rule(rule)

    def add_sequence(self, method: str, path: str, responses: List[Any]):
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.SEQUENCE,
            sequence=responses
        )
        self.add_rule(rule)

    def add_callback(self, method: str, path: str, callback: Callable):
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.CALLBACK,
            callback=callback
        )
        self.add_rule(rule)

    def add_chaos(self, method: str, path: str, chaos_config: Dict):
        """Inject failures: latency, errors, corruption."""
        rule = MockRule(
            method=method.upper(),
            path_pattern=path,
            strategy=MockStrategy.CHAOS,
            chaos_config=chaos_config
        )
        self.add_rule(rule)

    async def handle_request(self, request: web.Request):
        # Log request
        body = await request.read()
        request_data = {
            "id": str(uuid.uuid4())[:8],
            "method": request.method,
            "path": request.path,
            "headers": dict(request.headers),
            "query": dict(request.query),
            "body": body.decode() if body else "",
            "timestamp": asyncio.get_event_loop().time()
        }
        self.request_log.append(request_data)

        # Find matching rule
        rule = self._match_rule(request)

        if not rule:
            return web.Response(status=404, text="No mock rule matched")

        rule.times_matched += 1

        # Execute strategy
        return await self._execute_strategy(rule, request_data)

    def _match_rule(self, request: web.Request) -> Optional[MockRule]:
        for rule in self.rules:
            if not rule.active:
                continue
            if rule.method != request.method and rule.method != "*":
                continue
            if not self._match_path(rule.path_pattern, request.path):
                continue
            if not self._match_conditions(rule.conditions, request):
                continue
            return rule
        return None

    def _match_path(self, pattern: str, path: str) -> bool:
        # Simple glob matching
        import fnmatch
        return fnmatch.fnmatch(path, pattern)

    def _match_conditions(self, conditions: Dict, request: web.Request) -> bool:
        if "headers" in conditions:
            for k, v in conditions["headers"].items():
                if request.headers.get(k) != v:
                    return False
        if "query" in conditions:
            for k, v in conditions["query"].items():
                if request.query.get(k) != v:
                    return False
        return True

    async def _execute_strategy(self, rule: MockRule, request_data: Dict) -> web.Response:
        if rule.strategy == MockStrategy.STATIC:
            resp = rule.response
            return web.Response(
                status=resp.get("status", 200),
                body=json.dumps(resp["body"]) if isinstance(resp["body"], dict) else str(resp["body"]),
                headers=resp.get("headers", {"Content-Type": "application/json"})
            )

        elif rule.strategy == MockStrategy.SEQUENCE:
            idx = (rule.times_matched - 1) % len(rule.sequence)
            resp = rule.sequence[idx]
            return web.Response(
                status=resp.get("status", 200),
                body=json.dumps(resp["body"]) if isinstance(resp["body"], dict) else str(resp["body"]),
                headers=resp.get("headers", {"Content-Type": "application/json"})
            )

        elif rule.strategy == MockStrategy.CALLBACK:
            if asyncio.iscoroutinefunction(rule.callback):
                return await rule.callback(request_data)
            return rule.callback(request_data)

        elif rule.strategy == MockStrategy.CHAOS:
            return await self._apply_chaos(rule.chaos_config)

        return web.Response(status=500, text="Unknown strategy")

    async def _apply_chaos(self, config: Dict) -> web.Response:
        import random

        # Latency injection
        if "latency_ms" in config:
            await asyncio.sleep(config["latency_ms"] / 1000)

        # Error injection
        if "error_rate" in config and random.random() < config["error_rate"]:
            error_type = config.get("error_type", "500")
            if error_type == "timeout":
                raise asyncio.TimeoutError()
            return web.Response(status=int(error_type), text="Chaos error")

        # Response corruption
        if "corrupt" in config and random.random() < config["corrupt"]:
            return web.Response(
                status=200,
                body="{ invalid json",
                headers={"Content-Type": "application/json"}
            )

        # Default success
        return web.Response(
            status=200,
            body=json.dumps({"mock": True, "chaos": "applied"}),
            headers={"Content-Type": "application/json"}
        )

    async def start(self):
        runner = web.AppRunner(self.app)
        await runner.setup()
        site = web.TCPSite(runner, self.host, self.port)
        await site.start()
        print(f"Mock server running on http://{self.host}:{self.port}")
        return runner

    def get_request_log(self) -> List[Dict]:
        return self.request_log

    def reset_log(self):
        self.request_log.clear()

    def verify_call(self, method: str, path: str, times: int = 1) -> bool:
        count = sum(1 for r in self.request_log 
                   if r["method"] == method.upper() and self._match_path(path, r["path"]))
        return count == times

LLM API Mocking

# mocks/llm_mocks.py
from mocks.server import MockServer, MockStrategy, MockRule
from typing: Dict, List, Any
import json
import random

class LLMMockBuilder:
    """Build realistic LLM API mocks."""

    @staticmethod
    def openai_chat_completions(mock_server: MockServer, responses: List[Dict] = None):
        """Mock OpenAI /v1/chat/completions endpoint."""

        default_responses = [
            {
                "id": "chatcmpl-mock1",
                "object": "chat.completion",
                "created": 1699999999,
                "model": "gpt-4o-mini",
                "choices": [{
                    "index": 0,
                    "message": {
                        "role": "assistant",
                        "content": "This is a mocked response from APIMock-AI."
                    },
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30
                }
            }
        ]

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/chat/completions",
            strategy=MockStrategy.SEQUENCE,
            sequence=[{"status": 200, "body": r, "headers": {"Content-Type": "application/json"}} 
                     for r in (responses or default_responses)],
            priority=10
        ))

    @staticmethod
    def openai_embeddings(mock_server: MockServer, dimensions: int = 1536):
        """Mock OpenAI /v1/embeddings endpoint."""

        def embedding_callback(request_data):
            import json
            body = json.loads(request_data["body"]) if request_data["body"] else {}
            inputs = body.get("input", [])
            if isinstance(inputs, str):
                inputs = [inputs]

            embeddings = []
            for i, _ in enumerate(inputs):
                # Deterministic pseudo-embedding based on input hash
                import hashlib
                seed = hashlib.md5(inputs[i].encode()).hexdigest()
                random.seed(seed)
                emb = [random.uniform(-1, 1) for _ in range(dimensions)]
                # Normalize
                norm = sum(x*x for x in emb) ** 0.5
                emb = [x/norm for x in emb]
                embeddings.append({
                    "object": "embedding",
                    "embedding": emb,
                    "index": i
                })

            return web.Response(
                status=200,
                body=json.dumps({
                    "object": "list",
                    "data": embeddings,
                    "model": "text-embedding-3-small",
                    "usage": {"prompt_tokens": sum(len(x.split()) for x in inputs), "total_tokens": 100}
                }),
                headers={"Content-Type": "application/json"}
            )

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/embeddings",
            strategy=MockStrategy.CALLBACK,
            callback=embedding_callback,
            priority=10
        ))

    @staticmethod
    def anthropic_messages(mock_server: MockServer):
        """Mock Anthropic /v1/messages endpoint."""

        def anthropic_callback(request_data):
            import json
            body = json.loads(request_data["body"]) if request_data["body"] else {}

            return web.Response(
                status=200,
                body=json.dumps({
                    "id": "msg_mock1",
                    "type": "message",
                    "role": "assistant",
                    "content": [{"type": "text", "text": "Mocked Anthropic response"}],
                    "model": body.get("model", "claude-3-sonnet-20240229"),
                    "stop_reason": "end_turn",
                    "usage": {"input_tokens": 10, "output_tokens": 20}
                }),
                headers={"Content-Type": "application/json"}
            )

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/messages",
            strategy=MockStrategy.CALLBACK,
            callback=anthropic_callback,
            priority=10
        ))

    @staticmethod
    def add_rate_limit_mock(mock_server: MockServer, limit: int = 10, window: int = 60):
        """Add rate limiting simulation."""
        from collections import defaultdict
        import time

        request_times = defaultdict(list)

        def rate_limit_middleware(request_data):
            client_ip = request_data["headers"].get("x-forwarded-for", "unknown")
            now = time.time()

            # Clean old requests
            request_times[client_ip] = [t for t in request_times[client_ip] if now - t < window]

            if len(request_times[client_ip]) >= limit:
                return web.Response(
                    status=429,
                    body=json.dumps({
                        "error": {
                            "message": "Rate limit exceeded",
                            "type": "rate_limit_error",
                            "code": "rate_limit_exceeded"
                        }
                    }),
                    headers={
                        "Content-Type": "application/json",
                        "Retry-After": str(window)
                    }
                )

            request_times[client_ip].append(now)
            return None  # Continue to next rule

        mock_server.add_rule(MockRule(
            method="*",
            path_pattern="/v1/*",
            strategy=MockStrategy.CALLBACK,
            callback=rate_limit_middleware,
            priority=100  # Highest priority
        ))

    @staticmethod
    def add_streaming_mock(mock_server: MockServer):
        """Mock streaming responses."""

        async def streaming_callback(request_data):
            from aiohttp import web
            import asyncio

            response = web.StreamResponse(
                status=200,
                headers={
                    "Content-Type": "text/event-stream",
                    "Cache-Control": "no-cache",
                    "Connection": "keep-alive"
                }
            )
            await response.prepare(request_data.get("_aiohttp_request"))  # Would need actual request

            chunks = [
                'data: {"id":"mock","choices":[{"delta":{"content":"Hello"}}]}\n\n',
                'data: {"id":"mock","choices":[{"delta":{"content":" world"}}]}\n\n',
                'data: {"id":"mock","choices":[{"delta":{},"finish_reason":"stop"}]}\n\n',
                'data: [DONE]\n\n'
            ]

            for chunk in chunks:
                await response.write(chunk.encode())
                await asyncio.sleep(0.1)

            await response.write_eof()
            return response

        mock_server.add_rule(MockRule(
            method="POST",
            path_pattern="/v1/chat/completions",
            strategy=MockStrategy.CALLBACK,
            callback=streaming_callback,
            conditions={"headers": {"accept": "text/event-stream"}},
            priority=20
        ))

Part 2: Contract Testing

Contract Definition

# mocks/contracts.py
from dataclasses import dataclass, field
from typing: Dict, List, Optional, Any
from pydantic import BaseModel, Field
import json

class ParameterLocation(Enum):
    PATH = "path"
    QUERY = "query"
    HEADER = "header"
    BODY = "body"

@dataclass
class ContractParameter:
    name: str
    location: ParameterLocation
    required: bool = True
    schema: Dict = field(default_factory=dict)
    description: str = ""

@dataclass
class ContractResponse:
    status_code: int
    headers: Dict[str, str] = field(default_factory=dict)
    body_schema: Dict = field(default_factory=dict)
    description: str = ""

@dataclass
class APIContract:
    provider: str
    version: str
    base_url: str
    endpoints: Dict[str, Dict] = field(default_factory=dict)

    def add_endpoint(
        self,
        method: str,
        path: str,
        summary: str,
        parameters: List[ContractParameter] = None,
        responses: Dict[int, ContractResponse] = None,
        request_body: Dict = None
    ):
        key = f"{method.upper()} {path}"
        self.endpoints[key] = {
            "summary": summary,
            "parameters": parameters or [],
            "responses": responses or {},
            "request_body": request_body
        }

    def to_openapi(self) -> Dict:
        """Convert to OpenAPI 3.0 spec."""
        paths = {}
        for key, endpoint in self.endpoints.items():
            method, path = key.split(" ", 1)
            if path not in paths:
                paths[path] = {}

            parameters = []
            for p in endpoint["parameters"]:
                parameters.append({
                    "name": p.name,
                    "in": p.location.value,
                    "required": p.required,
                    "schema": p.schema,
                    "description": p.description
                })

            responses = {}
            for status, resp in endpoint["responses"].items():
                responses[str(status)] = {
                    "description": resp.description,
                    "headers": resp.headers,
                    "content": {
                        "application/json": {
                            "schema": resp.body_schema
                        }
                    }
                }

            paths[path][method.lower()] = {
                "summary": endpoint["summary"],
                "parameters": parameters,
                "responses": responses,
                "requestBody": endpoint.get("request_body", {})
            }

        return {
            "openapi": "3.0.0",
            "info": {"title": self.provider, "version": self.version},
            "servers": [{"url": self.base_url}],
            "paths": paths
        }

# OpenAI Contract
OPENAI_CONTRACT = APIContract(
    provider="OpenAI",
    version="2024-10-01-preview",
    base_url="https://api.openai.com/v1"
)

OPENAI_CONTRACT.add_endpoint(
    method="POST",
    path="/chat/completions",
    summary="Create chat completion",
    parameters=[
        ContractParameter("Authorization", ParameterLocation.HEADER, True, 
                         {"type": "string"}, "Bearer token"),
    ],
    request_body={
        "required": ["model", "messages"],
        "properties": {
            "model": {"type": "string"},
            "messages": {"type": "array", "items": {"type": "object"}},
            "temperature": {"type": "number", "minimum": 0, "maximum": 2},
            "max_tokens": {"type": "integer", "minimum": 1},
            "stream": {"type": "boolean"},
        }
    },
    responses={
        200: ContractResponse(200, {"Content-Type": "application/json"}, {
            "type": "object",
            "properties": {
                "id": {"type": "string"},
                "object": {"type": "string"},
                "created": {"type": "integer"},
                "model": {"type": "string"},
                "choices": {"type": "array"},
                "usage": {"type": "object"}
            }
        }, "Successful response"),
        400: ContractResponse(400, {}, {}, "Bad request"),
        401: ContractResponse(401, {}, {}, "Unauthorized"),
        429: ContractResponse(429, {}, {}, "Rate limited"),
    }
)

OPENAI_CONTRACT.add_endpoint(
    method="POST",
    path="/embeddings",
    summary="Create embeddings",
    request_body={
        "required": ["model", "input"],
        "properties": {
            "model": {"type": "string"},
            "input": {"type": ["string", "array"]},
            "dimensions": {"type": "integer"},
        }
    },
    responses={
        200: ContractResponse(200, {"Content-Type": "application/json"}, {
            "type": "object",
            "properties": {
                "object": {"type": "string"},
                "data": {"type": "array"},
                "model": {"type": "string"},
                "usage": {"type": "object"}
            }
        }, "Successful response"),
    }
)

Contract Testing

# mocks/contract_testing.py
from typing: Dict, List, Any
from dataclasses import dataclass
import jsonschema
import requests

@dataclass
class ContractTestResult:
    endpoint: str
    passed: bool
    errors: List[str]
    request: Dict
    response: Dict

class ContractTester:
    def __init__(self, contract: APIContract, base_url: str):
        self.contract = contract
        self.base_url = base_url.rstrip("/")

    def test_endpoint(
        self,
        method: str,
        path: str,
        request_data: Dict,
        expected_status: int = 200
    ) -> ContractTestResult:
        """Test single endpoint against contract."""
        endpoint_key = f"{method.upper()} {path}"
        contract_endpoint = self.contract.endpoints.get(endpoint_key)

        if not contract_endpoint:
            return ContractTestResult(
                endpoint=endpoint_key,
                passed=False,
                errors=[f"No contract defined for {endpoint_key}"],
                request=request_data,
                response={}
            )

        errors = []

        # Validate request
        if "request_body" in contract_endpoint:
            try:
                jsonschema.validate(request_data.get("json", {}), contract_endpoint["request_body"])
            except jsonschema.ValidationError as e:
                errors.append(f"Request validation failed: {e.message}")

        # Make request
        url = f"{self.base_url}{path}"
        headers = request_data.get("headers", {})
        json_body = request_data.get("json")

        try:
            response = requests.request(method, url, headers=headers, json=json_body, timeout=30)
        except Exception as e:
            return ContractTestResult(
                endpoint=endpoint_key,
                passed=False,
                errors=[f"Request failed: {e}"],
                request=request_data,
                response={}
            )

        # Validate response status
        expected_responses = contract_endpoint.get("responses", {})
        if expected_status not in expected_responses:
            # Check if any 2xx expected
            success_codes = [c for c in expected_responses if 200 <= c < 300]
            if response.status_code not in success_codes:
                errors.append(f"Unexpected status: {response.status_code}")

        # Validate response body
        if response.status_code in expected_responses:
            expected_schema = expected_responses[response.status_code].body_schema
            if expected_schema:
                try:
                    jsonschema.validate(response.json(), expected_schema)
                except jsonschema.ValidationError as e:
                    errors.append(f"Response validation failed: {e.message}")

        return ContractTestResult(
            endpoint=endpoint_key,
            passed=len(errors) == 0,
            errors=errors,
            request=request_data,
            response={"status": response.status_code, "body": response.json() if response.content else {}}
        )

    def run_all_tests(self, test_cases: List[Dict]) -> List[ContractTestResult]:
        """Run multiple contract tests."""
        results = []
        for tc in test_cases:
            result = self.test_endpoint(
                tc["method"],
                tc["path"],
                tc.get("request", {}),
                tc.get("expected_status", 200)
            )
            results.append(result)
        return results

# Test cases for OpenAI
OPENAI_TEST_CASES = [
    {
        "method": "POST",
        "path": "/v1/chat/completions",
        "request": {
            "headers": {"Authorization": "Bearer test-key"},
            "json": {
                "model": "gpt-4o-mini",
                "messages": [{"role": "user", "content": "Hello"}],
                "temperature": 0.7
            }
        },
        "expected_status": 200
    },
    {
        "method": "POST",
        "path": "/v1/chat/completions",
        "request": {
            "headers": {"Authorization": "Bearer test-key"},
            "json": {
                "model": "gpt-4o-mini",
                "messages": []
            }
        },
        "expected_status": 400
    },
    {
        "method": "POST",
        "path": "/v1/embeddings",
        "request": {
            "headers": {"Authorization": "Bearer test-key"},
            "json": {
                "model": "text-embedding-3-small",
                "input": "Test text"
            }
        },
        "expected_status": 200
    }
]

Part 3: Chaos Engineering

Failure Injection

# mocks/chaos.py
from dataclasses import dataclass
from typing: Dict, Any, Callable
from enum import Enum
import random
import asyncio

class ChaosType(Enum):
    LATENCY = "latency"
    ERROR = "error"
    TIMEOUT = "timeout"
    CORRUPTION = "corruption"
    PARTIAL = "partial"
    THROTTLE = "throttle"

@dataclass
class ChaosExperiment:
    name: str
    target_service: str
    chaos_type: ChaosType
    intensity: float  # 0-1
    duration_seconds: int
    conditions: Dict = field(default_factory=dict)

class ChaosEngine:
    def __init__(self, mock_server):
        self.mock_server = mock_server
        self.active_experiments: Dict[str, ChaosExperiment] = {}
        self.results: List[Dict] = []

    def inject_latency(self, path_pattern: str, min_ms: int, max_ms: int, probability: float = 1.0):
        """Inject random latency."""
        config = {
            "type": ChaosType.LATENCY.value,
            "min_ms": min_ms,
            "max_ms": max_ms,
            "probability": probability
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_errors(self, path_pattern: str, error_rate: float, status_codes: List[int] = None):
        """Inject random errors."""
        config = {
            "type": ChaosType.ERROR.value,
            "error_rate": error_rate,
            "status_codes": status_codes or [500, 502, 503, 504]
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_timeouts(self, path_pattern: str, timeout_rate: float):
        """Inject request timeouts."""
        config = {
            "type": ChaosType.TIMEOUT.value,
            "timeout_rate": timeout_rate
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_corruption(self, path_pattern: str, corruption_rate: float):
        """Inject response corruption."""
        config = {
            "type": ChaosType.CORRUPTION.value,
            "corruption_rate": corruption_rate
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    def inject_throttle(self, path_pattern: str, rate_limit: int, window_seconds: int):
        """Inject rate limiting."""
        config = {
            "type": ChaosType.THROTTLE.value,
            "rate_limit": rate_limit,
            "window_seconds": window_seconds
        }
        self.mock_server.add_chaos("*", path_pattern, config)

    async def run_experiment(self, experiment: ChaosExperiment, test_fn: Callable) -> Dict:
        """Run chaos experiment with test function."""
        self.active_experiments[experiment.name] = experiment

        start_time = asyncio.get_event_loop().time()
        errors = 0
        successes = 0
        latencies = []

        try:
            while asyncio.get_event_loop().time() - start_time < experiment.duration_seconds:
                try:
                    result = await test_fn()
                    if result.get("success"):
                        successes += 1
                    else:
                        errors += 1
                    if "latency_ms" in result:
                        latencies.append(result["latency_ms"])
                except Exception as e:
                    errors += 1

                await asyncio.sleep(0.1)
        finally:
            del self.active_experiments[experiment.name]

        result = {
            "experiment": experiment.name,
            "duration": experiment.duration_seconds,
            "total_requests": successes + errors,
            "successes": successes,
            "errors": errors,
            "error_rate": errors / max(1, successes + errors),
            "avg_latency_ms": sum(latencies) / len(latencies) if latencies else 0,
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)] if latencies else 0
        }

        self.results.append(result)
        return result

Resilience Testing Patterns

# mocks/resilience.py
"""
Resilience testing patterns for AI applications:

1. RETRY WITH BACKOFF
2. CIRCUIT BREAKER
3. FALLBACK RESPONSES
4. BULKHEAD ISOLATION
5. TIMEOUT MANAGEMENT
"""

from dataclasses import dataclass
from typing: Callable, Any, Optional
import asyncio
import time
from enum import Enum

class CircuitState(Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"

@dataclass
class RetryConfig:
    max_attempts: int = 3
    base_delay: float = 1.0
    max_delay: float = 30.0
    exponential_base: float = 2.0
    jitter: bool = True
    retry_on: tuple = (Exception,)

@dataclass
class CircuitBreakerConfig:
    failure_threshold: int = 5
    success_threshold: int = 2
    timeout: float = 30.0

class ResilientLLMClient:
    def __init__(
        self,
        base_client,
        retry_config: RetryConfig = None,
        circuit_config: CircuitBreakerConfig = None,
        fallback: Callable = None
    ):
        self.client = base_client
        self.retry_config = retry_config or RetryConfig()
        self.circuit_config = circuit_config or CircuitBreakerConfig()
        self.fallback = fallback

        self.circuit_state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0

    async def chat_completion(self, **kwargs) -> Dict:
        return await self._execute_with_resilience("chat_completion", kwargs)

    async def _execute_with_resilience(self, method: str, kwargs: Dict) -> Dict:
        # Circuit breaker check
        if self.circuit_state == CircuitState.OPEN:
            if time.time() - self.last_failure_time > self.circuit_config.timeout:
                self.circuit_state = CircuitState.HALF_OPEN
            else:
                if self.fallback:
                    return await self.fallback(method, kwargs)
                raise Exception("Circuit breaker open")

        # Retry loop
        last_exception = None
        for attempt in range(self.retry_config.max_attempts):
            try:
                result = await getattr(self.client, method)(**kwargs)

                # Success - reset circuit breaker
                self._on_success()
                return result

            except self.retry_config.retry_on as e:
                last_exception = e
                self._on_failure()

                if attempt < self.retry_config.max_attempts - 1:
                    delay = self._calculate_delay(attempt)
                    await asyncio.sleep(delay)

        # All retries failed
        if self.fallback:
            return await self.fallback(method, kwargs)
        raise last_exception

    def _calculate_delay(self, attempt: int) -> float:
        delay = min(
            self.retry_config.base_delay * (self.retry_config.exponential_base ** attempt),
            self.retry_config.max_delay
        )
        if self.retry_config.jitter:
            import random
            delay *= (0.5 + random.random())  # 50-150% of delay
        return delay

    def _on_success(self):
        self.failure_count = 0
        if self.circuit_state == CircuitState.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.circuit_config.success_threshold:
                self.circuit_state = CircuitState.CLOSED
                self.success_count = 0

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        self.success_count = 0

        if self.failure_count >= self.circuit_config.failure_threshold:
            self.circuit_state = CircuitState.OPEN

# Fallback implementations
async def cached_fallback(method: str, kwargs: Dict) -> Dict:
    """Return cached response."""
    return {"fallback": True, "source": "cache", "data": "cached response"}

async def static_fallback(method: str, kwargs: Dict) -> Dict:
    """Return static response."""
    return {"fallback": True, "source": "static", "data": "I'm currently experiencing issues."}

async def degraded_fallback(method: str, kwargs: Dict) -> Dict:
    """Return degraded but functional response."""
    # Use smaller model, fewer tokens, no streaming
    kwargs["model"] = "gpt-3.5-turbo"
    kwargs["max_tokens"] = 100
    kwargs["stream"] = False
    return {"fallback": True, "source": "degraded", **kwargs}

Part 4: Testing Strategies

Unit Testing with Mocks

# tests/test_with_mocks.py
import pytest
from unittest.mock import AsyncMock, patch
from mocks.server import MockServer
from mocks.llm_mocks import LLMMockBuilder

class TestAIApplication:
    @pytest.fixture
    async def mock_server(self):
        server = MockServer("localhost", 8081)
        LLMMockBuilder.openai_chat_completions(server)
        LLMMockBuilder.openai_embeddings(server)
        runner = await server.start()
        yield server
        await runner.cleanup()

    @pytest.fixture
    def ai_client(self, mock_server):
        from myapp import AIClient
        return AIClient(base_url="http://localhost:8081")

    @pytest.mark.asyncio
    async def test_chat_completion(self, ai_client, mock_server):
        response = await ai_client.chat(
            messages=[{"role": "user", "content": "Hello"}],
            model="gpt-4o-mini"
        )

        assert response["choices"][0]["message"]["content"] == "This is a mocked response from APIMock-AI."
        assert mock_server.verify_call("POST", "/v1/chat/completions", times=1)

    @pytest.mark.asyncio
    async def test_embeddings(self, ai_client, mock_server):
        embeddings = await ai_client.embed(["text1", "text2"])

        assert len(embeddings) == 2
        assert len(embeddings[0]) == 1536
        assert mock_server.verify_call("POST", "/v1/embeddings", times=1)

    @pytest.mark.asyncio
    async def test_rate_limit_handling(self, ai_client, mock_server):
        # Add rate limit mock
        LLMMockBuilder.add_rate_limit_mock(mock_server, limit=2, window=60)

        # First two should succeed
        await ai_client.chat(messages=[{"role": "user", "content": "1"}])
        await ai_client.chat(messages=[{"role": "user", "content": "2"}])

        # Third should handle rate limit
        with pytest.raises(Exception) as exc_info:
            await ai_client.chat(messages=[{"role": "user", "content": "3"}])

        assert "rate limit" in str(exc_info.value).lower()

Integration Testing

# tests/integration_test.py
import pytest
from mocks.server import MockServer
from mocks.llm_mocks import LLMMockBuilder
from mocks.chaos import ChaosEngine, ChaosType

class TestResilience:
    @pytest.fixture
    async def chaos_server(self):
        server = MockServer("localhost", 8082)
        LLMMockBuilder.openai_chat_completions(server)
        runner = await server.start()
        yield server, ChaosEngine(server)
        await runner.cleanup()

    @pytest.mark.asyncio
    async def test_latency_resilience(self, chaos_server):
        server, chaos = chaos_server

        # Inject 1-2s latency
        chaos.inject_latency("/v1/chat/completions", 1000, 2000, probability=1.0)

        from myapp import ResilientAIClient
        client = ResilientAIClient(base_url="http://localhost:8082", timeout=5.0)

        # Should succeed with retry
        response = await client.chat(messages=[{"role": "user", "content": "Test"}])
        assert "fallback" not in response or response["fallback"] == False

    @pytest.mark.asyncio
    async def test_error_resilience(self, chaos_server):
        server, chaos = chaos_server

        # Inject 50% error rate
        chaos.inject_errors("/v1/chat/completions", 0.5, [500, 503])

        from myapp import ResilientAIClient
        client = ResilientAIClient(
            base_url="http://localhost:8082",
            fallback=static_fallback
        )

        # Should eventually succeed or fallback
        response = await client.chat(messages=[{"role": "user", "content": "Test"}])
        assert response is not None

    @pytest.mark.asyncio
    async def test_circuit_breaker(self, chaos_server):
        server, chaos = chaos_server

        # Inject 100% errors
        chaos.inject_errors("/v1/chat/completions", 1.0, [500])

        from myapp import ResilientAIClient
        client = ResilientAIClient(
            base_url="http://localhost:8082",
            fallback=cached_fallback
        )

        # First few calls fail, then circuit opens
        for _ in range(3):
            try:
                await client.chat(messages=[{"role": "user", "content": "Test"}])
            except:
                pass

        # Next call should use fallback immediately
        response = await client.chat(messages=[{"role": "user", "content": "Test"}])
        assert response.get("fallback") == True
        assert response.get("source") == "cache"

Part 5: CI/CD Integration

GitHub Actions Workflow

# .github/workflows/api-mock-tests.yml
name: API Mock Tests

on:
  push:
    branches: [main, develop]
  pull_request:
  schedule:
    - cron: '0 6 * * 1'  # Weekly Monday 6 AM

jobs:
  contract-tests:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Start Mock Server
        run: |
          python -m mocks.server --port 8080 &
          sleep 3

      - name: Configure OpenAI Mocks
        run: python -m mocks.configure_openai_mocks

      - name: Run Contract Tests
        run: pytest tests/contract_tests.py -v

      - name: Run Unit Tests
        run: pytest tests/test_with_mocks.py -v

      - name: Run Resilience Tests
        run: pytest tests/integration_test.py -v

      - name: Generate Report
        run: python -m mocks.generate_report > mock-test-report.json

      - name: Upload Report
        uses: actions/upload-artifact@v4
        with:
          name: mock-test-report
          path: mock-test-report.json

  chaos-engineering:
    runs-on: ubuntu-latest
    if: github.event_name == 'schedule'
    steps:
      - uses: actions/checkout@v4

      - name: Run Chaos Experiments
        run: |
          python -m mocks.run_chaos_experiments \
            --experiment latency \
            --experiment errors \
            --experiment timeouts \
            --duration 60

      - name: Analyze Results
        run: python -m mocks.analyze_chaos_results

      - name: Alert on Degradation
        if: failure()
        uses: slackapi/slack-github-action@v1.23.0
        with:
          payload: |
            {
              "text": "Chaos engineering detected resilience degradation",
              "blocks": [...]
            }

Local Development Setup

# scripts/start_mocks.sh
#!/bin/bash

# Start mock servers for local development
echo "Starting APIMock-AI servers..."

# LLM API mock
python -m mocks.server --port 8080 --config mocks/configs/openai.yaml &

# Vector DB mock
python -m mocks.server --port 8081 --config mocks/configs/qdrant.yaml &

# Embedding service mock
python -m mocks.server --port 8082 --config mocks/configs/embeddings.yaml &

echo "Mock servers running:"
echo "  OpenAI API: http://localhost:8080"
echo "  Qdrant: http://localhost:8081"
echo "  Embeddings: http://localhost:8082"

# Save PIDs for cleanup
jobs -p > .mock_pids

# mocks/configs/openai.yaml
server:
  host: localhost
  port: 8080

rules:
  - method: POST
    path: /v1/chat/completions
    strategy: sequence
    responses:
      - status: 200
        body:
          id: "chatcmpl-dev"
          object: "chat.completion"
          choices:
            - index: 0
              message:
                role: assistant
                content: "Development mock response"
              finish_reason: stop
          usage:
            prompt_tokens: 10
            completion_tokens: 20
            total_tokens: 30

  - method: POST
    path: /v1/embeddings
    strategy: callback
    module: mocks.llm_mocks
    function: openai_embeddings_callback

Part 6: Production Patterns

Mock Recording/Playback

# mocks/recorder.py
"""Record real API calls and replay as mocks."""

from dataclasses import dataclass, field
from typing: Dict, List, Any
import json
import hashlib
from datetime import datetime

@dataclass
class RecordedInteraction:
    id: str
    timestamp: str
    request: Dict
    response: Dict
    metadata: Dict = field(default_factory=dict)

class MockRecorder:
    def __init__(self, storage_path: str = "mocks/recordings"):
        self.storage_path = storage_path
        self.recordings: List[RecordedInteraction] = []
        self.recording = False

    def start_recording(self):
        self.recording = True
        self.recordings = []

    def stop_recording(self) -> List[RecordedInteraction]:
        self.recording = False
        return self.recordings

    def record(self, request: Dict, response: Dict, metadata: Dict = None):
        if not self.recording:
            return

        interaction = RecordedInteraction(
            id=hashlib.md5(f"{request['method']}{request['path']}{datetime.now()}".encode()).hexdigest()[:12],
            timestamp=datetime.utcnow().isoformat(),
            request=request,
            response=response,
            metadata=metadata or {}
        )
        self.recordings.append(interaction)

    def save(self, name: str):
        import os
        os.makedirs(self.storage_path, exist_ok=True)

        filepath = f"{self.storage_path}/{name}.json"
        with open(filepath, "w") as f:
            json.dump([{
                "id": r.id,
                "timestamp": r.timestamp,
                "request": r.request,
                "response": r.response,
                "metadata": r.metadata
            } for r in self.recordings], f, indent=2)

    def load(self, name: str) -> List[RecordedInteraction]:
        filepath = f"{self.storage_path}/{name}.json"
        with open(filepath) as f:
            data = json.load(f)
        return [RecordedInteraction(**r) for r in data]

    def generate_mock_rules(self, recordings: List[RecordedInteraction]) -> List[MockRule]:
        """Generate mock rules from recordings."""
        rules = []

        # Group by endpoint
        by_endpoint = {}
        for r in recordings:
            key = f"{r.request['method']} {r.request['path']}"
            if key not in by_endpoint:
                by_endpoint[key] = []
            by_endpoint[key].append(r)

        for endpoint, interactions in by_endpoint.items():
            method, path = endpoint.split(" ", 1)

            if len(interactions) == 1:
                # Single response - static
                r = interactions[0]
                rules.append(MockRule(
                    method=method,
                    path_pattern=path,
                    strategy=MockStrategy.STATIC,
                    response={
                        "status": r.response["status"],
                        "body": r.response["body"],
                        "headers": r.response.get("headers", {})
                    }
                ))
            else:
                # Multiple responses - sequence
                rules.append(MockRule(
                    method=method,
                    path_pattern=path,
                    strategy=MockStrategy.SEQUENCE,
                    sequence=[{
                        "status": r.response["status"],
                        "body": r.response["body"],
                        "headers": r.response.get("headers", {})
                    } for r in interactions]
                ))

        return rules

Part 7: APIMock-AI Product Features

Key Features Checklist

CORE MOCKING
✅ Static responses
✅ Sequential responses
✅ Template-based responses
✅ Dynamic callbacks
✅ Proxy mode
✅ Chaos engineering

CONTRACT TESTING
✅ OpenAPI 3.0 support
✅ Request validation
✅ Response validation
✅ Schema evolution detection
✅ Breaking change alerts

DEVELOPER EXPERIENCE
✅ Zero-config startup
✅ Hot reload
✅ Request logging
✅ Verification API
✅ VS Code extension
✅ CLI tool

RESILIENCE TESTING
✅ Latency injection
✅ Error injection
✅ Timeout simulation
✅ Response corruption
✅ Rate limiting
✅ Circuit breaker testing

INTEGRATIONS
✅ OpenAI API
✅ Anthropic API
✅ Google Gemini API
✅ Vector databases (Qdrant, Pinecone, Weaviate, Milvus)
✅ Custom HTTP APIs
✅ GraphQL endpoints
✅ gRPC services

DEPLOYMENT
✅ Docker image
✅ Kubernetes operator
✅ Helm chart
✅ GitHub Action
✅ GitLab CI
✅ Local development

Resources

• APIMock-AI Documentation
• WireMock
• MockServer
• Pact Contract Testing
• Chaos Engineering Principles
• Circuit Breaker Pattern
• Testcontainers

---

API mocking is not a testing luxury — it's a development necessity. The teams that mock external dependencies from day one ship faster, debug easier, and sleep better. APIMock-AI makes it trivial to simulate the entire AI ecosystem locally.