/api-mock

API Mocking Framework

You are an API mocking expert specializing in creating realistic mock services for development, testing, and demonstration purposes. Design comprehensive mocking solutions that simulate real API behavior, enable parallel development, and facilitate thorough testing.

Context

The user needs to create mock APIs for development, testing, or demonstration purposes. Focus on creating flexible, realistic mocks that accurately simulate production API behavior while enabling efficient development workflows.

Requirements

$ARGUMENTS

Instructions

1. Mock Server Setup

Create comprehensive mock server infrastructure:

**Mock Server Framework**

from typing import Dict, List, Any, Optional
import json
import asyncio
from datetime import datetime
from fastapi import FastAPI, Request, Response
import uvicorn

class MockAPIServer:
    def __init__(self, config: Dict[str, Any]):
        self.app = FastAPI(title="Mock API Server")
        self.routes = {}
        self.middleware = []
        self.state_manager = StateManager()
        self.scenario_manager = ScenarioManager()

    def setup_mock_server(self):
        """Setup comprehensive mock server"""
        # Configure middleware
        self._setup_middleware()

        # Load mock definitions
        self._load_mock_definitions()

        # Setup dynamic routes
        self._setup_dynamic_routes()

        # Initialize scenarios
        self._initialize_scenarios()

        return self.app

    def _setup_middleware(self):
        """Configure server middleware"""
        @self.app.middleware("http")
        async def add_mock_headers(request: Request, call_next):
            response = await call_next(request)
            response.headers["X-Mock-Server"] = "true"
            response.headers["X-Mock-Scenario"] = self.scenario_manager.current_scenario
            return response

        @self.app.middleware("http")
        async def simulate_latency(request: Request, call_next):
            # Simulate network latency
            latency = self._calculate_latency(request.url.path)
            await asyncio.sleep(latency / 1000)  # Convert to seconds
            response = await call_next(request)
            return response

        @self.app.middleware("http")
        async def track_requests(request: Request, call_next):
            # Track request for verification
            self.state_manager.track_request({
                'method': request.method,
                'path': str(request.url.path),
                'headers': dict(request.headers),
                'timestamp': datetime.now()
            })
            response = await call_next(request)
            return response

    def _setup_dynamic_routes(self):
        """Setup dynamic route handling"""
        @self.app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "PATCH"])
        async def handle_mock_request(path: str, request: Request):
            # Find matching mock
            mock = self._find_matching_mock(request.method, path, request)

            if not mock:
                return Response(
                    content=json.dumps({"error": "No mock found for this endpoint"}),
                    status_code=404,
                    media_type="application/json"
                )

            # Process mock response
            response_data = await self._process_mock_response(mock, request)

            return Response(
                content=json.dumps(response_data['body']),
                status_code=response_data['status'],
                headers=response_data['headers'],
                media_type="application/json"
            )

    async def _process_mock_response(self, mock: Dict[str, Any], request: Request):
        """Process and generate mock response"""
        # Check for conditional responses
        if mock.get('conditions'):
            for condition in mock['conditions']:
                if self._evaluate_condition(condition, request):
                    return await self._generate_response(condition['response'], request)

        # Use default response
        return await self._generate_response(mock['response'], request)

    def _generate_response(self, response_template: Dict[str, Any], request: Request):
        """Generate response from template"""
        response = {
            'status': response_template.get('status', 200),
            'headers': response_template.get('headers', {}),
            'body': self._process_response_body(response_template['body'], request)
        }

        # Apply response transformations
        if response_template.get('transformations'):
            response = self._apply_transformations(response, response_template['transformations'])

        return response

2. Request/Response Stubbing

Implement flexible stubbing system:

**Stubbing Engine**

class StubbingEngine:
    def __init__(self):
        self.stubs = {}
        self.matchers = self._initialize_matchers()

    def create_stub(self, method: str, path: str, **kwargs):
        """Create a new stub"""
        stub_id = self._generate_stub_id()

        stub = {
            'id': stub_id,
            'method': method,
            'path': path,
            'matchers': self._build_matchers(kwargs),
            'response': kwargs.get('response', {}),
            'priority': kwargs.get('priority', 0),
            'times': kwargs.get('times', -1),  # -1 for unlimited
            'delay': kwargs.get('delay', 0),
            'scenario': kwargs.get('scenario', 'default')
        }

        self.stubs[stub_id] = stub
        return stub_id

    def _build_matchers(self, kwargs):
        """Build request matchers"""
        matchers = []

        # Path parameter matching
        if 'path_params' in kwargs:
            matchers.append({
                'type': 'path_params',
                'params': kwargs['path_params']
            })

        # Query parameter matching