Initial upload: mockapi - OpenAPI Mock Server Generator

src/mockapi/core/delay.py

@@ -0,0 +1,108 @@
+"""Response delay middleware."""
+
+import random
+import time
+from typing import Callable, Optional
+
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.requests import Request
+from starlette.responses import Response
+
+
+class DelayMiddleware(BaseHTTPMiddleware):
+    """Middleware that adds artificial delays to responses."""
+
+    def __init__(
+        self,
+        app,
+        delay: int = 0,
+        random_delay: bool = False,
+        min_delay: int = 100,
+        max_delay: int = 2000,
+    ):
+        """Initialize the delay middleware.
+
+        Args:
+            app: The ASGI application
+            delay: Fixed delay in milliseconds
+            random_delay: Use random delays
+            min_delay: Minimum delay in ms (for random mode)
+            max_delay: Maximum delay in ms (for random mode)
+        """
+        super().__init__(app)
+        self.delay = delay
+        self.random_delay = random_delay
+        self.min_delay = min_delay
+        self.max_delay = max_delay
+
+    async def dispatch(self, request: Request, call_next: Callable) -> Response:
+        """Process the request with optional delay."""
+        if self.random_delay:
+            actual_delay = random.randint(self.min_delay, self.max_delay) / 1000.0
+        else:
+            actual_delay = self.delay / 1000.0
+
+        if actual_delay > 0:
+            time.sleep(actual_delay)
+
+        response = await call_next(request)
+        return response
+
+    @staticmethod
+    def wrap(app, delay: int = 0, random_delay: bool = False) -> "DelayMiddleware":
+        """Wrap an app with delay middleware.
+
+        Args:
+            app: The application to wrap
+            delay: Fixed delay in milliseconds
+            random_delay: Use random delays
+
+        Returns:
+            Wrapped application
+        """
+        return DelayMiddleware(app, delay=delay, random_delay=random_delay)
+
+
+class ErrorSimulator:
+    """Simulates error responses for testing."""
+
+    def __init__(
+        self,
+        error_probability: float = 0.0,
+        default_error_code: int = 500,
+    ):
+        """Initialize the error simulator.
+
+        Args:
+            error_probability: Probability of returning an error (0.0 to 1.0)
+            default_error_code: Default HTTP error code
+        """
+        self.error_probability = error_probability
+        self.default_error_code = default_error_code
+
+    def should_return_error(self) -> bool:
+        """Determine if an error should be returned.
+
+        Returns:
+            True if error should be returned
+        """
+        return random.random() < self.error_probability
+
+    def get_error_response(
+        self,
+        status_code: Optional[int] = None,
+        message: Optional[str] = None,
+    ) -> tuple:
+        """Get an error response tuple.
+
+        Args:
+            status_code: HTTP status code
+            message: Error message
+
+        Returns:
+            Tuple of (body_dict, status_code)
+        """
+        return (
+            {"error": message or "Simulated error"},
+            status_code or self.default_error_code,
+        )