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.

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.

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 Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 )) Enter fullscreen mode Exit fullscreen mode 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 " ), } ) Enter fullscreen mode Exit fullscreen mode 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 } ] Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 } Enter fullscreen mode Exit fullscreen mode 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 () Enter fullscreen mode Exit fullscreen mode 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 " Enter fullscreen mode Exit fullscreen mode 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": } Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode # 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 Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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.