Initial upload: mockapi - OpenAPI Mock Server Generator

src/mockapi/core/server_generator.py

@@ -0,0 +1,155 @@
+"""Mock Server Generator using connexion."""
+
+import random
+from typing import Any, Dict, Optional
+
+import connexion
+from connexion import App, Resolution
+from connexion.resolver import Resolver
+
+from mockapi.core.config import Config
+from mockapi.core.generators import DataGenerator
+from mockapi.core.delay import DelayMiddleware
+
+
+class MockOperationResolver(Resolver):
+    """Custom operation resolver for mock API."""
+
+    def __init__(self, mock_server_generator):
+        """Initialize the resolver.
+
+        Args:
+            mock_server_generator: The MockServerGenerator instance
+        """
+        super().__init__()
+        self.mock_server_generator = mock_server_generator
+
+    def resolve(self, operation):
+        """Resolve an operation to a mock function.
+
+        Args:
+            operation: The operation object from connexion
+
+        Returns:
+            Resolution object with the mock function
+        """
+        operation_id = self.resolve_operation_id(operation)
+        for path, path_item in self.mock_server_generator.spec.get("paths", {}).items():
+            for method, op_def in path_item.items():
+                if method in ["get", "post", "put", "delete", "patch", "options", "head"]:
+                    if op_def.get("operationId") == operation_id:
+                        func = self.mock_server_generator._create_mock_function(op_def)
+                        return Resolution(func, operation_id)
+        raise Exception(f"Operation {operation_id} not found")
+
+
+class MockServerGenerator:
+    """Generates a mock server from an OpenAPI specification."""
+
+    def __init__(self, spec: Dict[str, Any], config: Optional[Config] = None):
+        """Initialize the mock server generator.
+
+        Args:
+            spec: The OpenAPI specification dictionary
+            config: Configuration object
+        """
+        self.spec = spec
+        self.config = config or Config()
+        schemas = spec.get("components", {}).get("schemas", {})
+        self.data_generator = DataGenerator(seed=self.config.seed, schemas=schemas)
+        self.app: Optional[App] = None
+
+    def generate(self) -> App:
+        """Generate the connexion application.
+
+        Returns:
+            Configured connexion App instance
+        """
+        self.app = connexion.App(__name__, specification_dir=".")
+
+        self.app.add_api(
+            self.spec,
+            validate_responses=self.config.validate_responses,
+            strict_validation=self.config.strict_validation,
+            resolver=MockOperationResolver(self),
+        )
+
+        if self.config.delay > 0 or self.config.random_delay:
+            self.app = DelayMiddleware.wrap(
+                self.app,
+                delay=self.config.delay,
+                random_delay=self.config.random_delay,
+            )
+
+        return self.app
+
+    def _create_mock_function(self, operation: Dict[str, Any]):
+        """Create a mock function for an operation.
+
+        Args:
+            operation: The OpenAPI operation definition
+
+        Returns:
+            Function that generates mock responses
+        """
+
+        def mock_function(*args, **kwargs):
+            mock_config = operation.get("x-mock-config", {})
+
+            if self._should_return_error(mock_config):
+                return self._generate_error_response(mock_config)
+
+            response_schema = operation.get("responses", {}).get("200", {}).get(
+                "content", {}
+            ).get("application/json", {}).get("schema")
+
+            if response_schema:
+                data = self.data_generator.generate(response_schema)
+                return data, 200
+
+            return {"message": "Mock response"}, 200
+
+        mock_function._mockapi_operation = operation
+        return mock_function
+
+    def _should_return_error(self, mock_config: Dict[str, Any]) -> bool:
+        """Determine if we should return an error response.
+
+        Args:
+            mock_config: x-mock-config extension data
+
+        Returns:
+            True if error should be returned
+        """
+        error_probability = mock_config.get("errorProbability", 0)
+        return random.random() < error_probability
+
+    def _generate_error_response(self, mock_config: Dict[str, Any]) -> tuple:
+        """Generate an error response.
+
+        Args:
+            mock_config: x-mock-config extension data
+
+        Returns:
+            Tuple of (response_body, status_code)
+        """
+        status_code = mock_config.get("errorCode", 500)
+        error_message = mock_config.get("errorMessage", "Mock error")
+        return {"error": error_message}, status_code
+
+
+def create_mock_server(
+    spec: Dict[str, Any],
+    config: Optional[Config] = None,
+) -> App:
+    """Create a mock server from an OpenAPI spec.
+
+    Args:
+        spec: OpenAPI specification dictionary
+        config: Configuration object
+
+    Returns:
+        Configured connexion App
+    """
+    generator = MockServerGenerator(spec, config)
+    return generator.generate()