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 

# 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 

# 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.Source：https://dev.to/ipsphil_digital/api-mocking-for-ai-applications-complete-guide-to-apimock-ai-patterns-4l98
My favorites