guidellm.entrypoints

class MockServer:
    """
    High-performance mock server implementing OpenAI and vLLM API endpoints.

    Provides a Sanic-based web server that simulates API responses with configurable
    timing characteristics for testing and benchmarking purposes. Supports chat
    completions, text completions, tokenization endpoints, and model listing with
    realistic latency patterns to enable comprehensive performance validation.

    Example:
    ::
        config = ServerConfig(model="test-model", port=8080)
        server = MockServer(config)
        server.run()
    """

    def __init__(self, config: MockServerConfig) -> None:
        """
        Initialize the mock server with configuration.

        :param config: Server configuration containing network settings and response
            timing parameters
        """
        self.config = config
        self.app = Sanic("guidellm-mock-server")
        self.chat_handler = ChatCompletionsHandler(config)
        self.completions_handler = CompletionsHandler(config)
        self.responses_handler = ResponsesHandler(config)
        self.tokenizer_handler = TokenizerHandler(config)
        # Deterministic test controls: count accepted generation requests and
        # optionally serialize them with a semaphore (lazy-created in the loop).
        self._accepted_generation_requests = 0
        self._concurrency_semaphore: asyncio.Semaphore | None = None

        self._setup_middleware()
        self._setup_routes()
        self._setup_error_handlers()

    def _get_concurrency_semaphore(self) -> asyncio.Semaphore | None:
        """Return the concurrency semaphore, creating it in the running loop."""
        if self.config.max_concurrent_requests is None:
            return None
        if self._concurrency_semaphore is None:
            self._concurrency_semaphore = asyncio.Semaphore(
                self.config.max_concurrent_requests
            )
        return self._concurrency_semaphore

    async def _run_generation(
        self,
        handler: Callable[[Request], Awaitable[HTTPResponse]],
        request: Request,
    ) -> HTTPResponse:
        """
        Run a generation handler with optional fail-after and concurrency limits.

        :param handler: Async generation endpoint handler
        :param request: Incoming Sanic request
        :return: Handler response, or HTTP 500 when fail_after_requests is exceeded
        """
        fail_after = self.config.fail_after_requests
        if fail_after is not None and self._accepted_generation_requests >= fail_after:
            return response.json(
                {
                    "error": {
                        "message": (
                            f"Mock server fail_after_requests={fail_after} exceeded"
                        ),
                        "type": "server_error",
                        "code": "fail_after_requests",
                    }
                },
                status=500,
            )

        self._accepted_generation_requests += 1
        semaphore = self._get_concurrency_semaphore()
        if semaphore is None:
            return await handler(request)
        async with semaphore:
            return await handler(request)

    def _audio_usage(self, file: File, text: str) -> dict[str, int | float]:
        """
        Build usage statistics for an audio endpoint response.

        Charges prompt tokens for the uploaded audio using the configured
        per-second rate applied to the duration estimated from the payload
        size, and counts completion tokens from the generated text.

        :param file: Uploaded audio file from the multipart form
        :param text: Text returned in the response body
        :return: Usage dict with prompt, completion, and total token counts
        """
        audio_seconds = estimate_audio_seconds(len(file.body), file.type or file.name)
        prompt_tokens = math.ceil(audio_seconds * self.config.audio_tokens_per_second)
        completion_tokens = len(self.tokenizer_handler.tokenizer.tokenize(text))
        return {
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
            "seconds": round(audio_seconds, 3),
        }

    def _setup_middleware(self):
        """Setup middleware for CORS, logging, etc."""

        @self.app.middleware("request")
        async def log_request_received(request: Request) -> None:
            """Log request arrival when log_request_received is enabled."""
            if self.config.log_request_received:
                logger.info(
                    "Request received: %s %s from %s",
                    request.method,
                    request.path,
                    request.ip,
                )

        @self.app.middleware("request")
        async def add_cors_headers(_request: Request) -> None:
            """Add CORS headers to all requests."""
            return None  # noqa: RET501

        @self.app.middleware("response")
        async def add_response_headers(
            _request: Any, resp: BaseHTTPResponse
        ) -> HTTPResponse:
            """Add standard response headers."""
            resp.headers["Access-Control-Allow-Origin"] = "*"
            resp.headers["Access-Control-Allow-Methods"] = "GET, POST, OPTIONS"
            resp.headers["Access-Control-Allow-Headers"] = "Content-Type, Authorization"
            resp.headers["Server"] = "guidellm-mock-server"
            return resp  # type: ignore[return-value]

    def _setup_routes(self):  # noqa: C901
        @self.app.get("/health")
        async def health_check(_request: Request):
            return response.json({"status": "healthy", "timestamp": time.time()})

        @self.app.get("/v1/models")
        async def list_models(_request: Request):
            return response.json(
                {
                    "object": "list",
                    "data": [
                        {
                            "id": self.config.model,
                            "object": "model",
                            "created": int(time.time()),
                            "owned_by": "guidellm-mock",
                        }
                    ],
                }
            )

        @self.app.route("/v1/chat/completions", methods=["POST", "OPTIONS"])
        async def chat_completions(request: Request):
            if request.method == "OPTIONS":
                return response.text("", status=204)
            return await self._run_generation(self.chat_handler.handle, request)

        @self.app.route("/v1/completions", methods=["POST", "OPTIONS"])
        async def completions(request: Request):
            if request.method == "OPTIONS":
                return response.text("", status=204)
            return await self._run_generation(self.completions_handler.handle, request)

        @self.app.route("/v1/responses", methods=["POST", "OPTIONS"])
        async def responses(request: Request):
            if request.method == "OPTIONS":
                return response.text("", status=204)
            return await self._run_generation(self.responses_handler.handle, request)

        @self.app.route("/tokenize", methods=["POST", "OPTIONS"])
        async def tokenize(request: Request):
            if request.method == "OPTIONS":
                return response.text("", status=204)
            return await self.tokenizer_handler.tokenize(request)

        @self.app.route("/detokenize", methods=["POST", "OPTIONS"])
        async def detokenize(request: Request):
            if request.method == "OPTIONS":
                return response.text("", status=204)
            return await self.tokenizer_handler.detokenize(request)

        @self.app.route("/v1/audio/transcriptions", methods=["POST", "OPTIONS"])
        async def audio_transcriptions(request: Request) -> HTTPResponse:
            """
            Mock OpenAI audio transcription endpoint:
            - receives multipart/form-data
            - file field contains audio file
            - model field is optional, default to "mock-model"
            - returns "transcribed text"
            """
            if request.method == "OPTIONS":
                return response.text("", status=204)
            if request.files is None or request.form is None:
                return response.json({"error": "No form data provided"}, status=400)
            file: File | None = request.files.get("file")
            if "file" not in request.files or "model" not in request.form:
                return response.json(
                    {"error": "Missing 'file' in form-data"}, status=400
                )

            file = cast("File", file)
            model = request.form.get("model", "mock-model")
            text = f"Mock transcription for {file.name}"

            return response.json(
                {
                    "text": text,
                    "file_size": len(file.body),
                    "model_used": model,
                    "transcription": f"Transcribed({file.name}) using {model}",
                    "usage": self._audio_usage(file, text),
                }
            )

        @self.app.route("/v1/audio/translations", methods=["POST", "OPTIONS"])
        async def audio_translations(request: Request) -> HTTPResponse:
            """
            Mock OpenAI audio translation endpoint:
            - receives multipart/form-data
            - file field contains audio file
            - model field is optional, default to "mock-model"
            - returns translated text
            """
            if request.method == "OPTIONS":
                return response.text("", status=204)
            if request.files is None or request.form is None:
                return response.json({"error": "No form data provided"}, status=400)
            file: File | None = request.files.get("file")
            if "file" not in request.files or "model" not in request.form:
                return response.json(
                    {"error": "Missing 'file' in form-data"}, status=400
                )

            file = cast("File", file)
            decoded_text = (
                "This is a mock translation result."  # mock output tranlated text
            )

            return response.json(
                {
                    "text": decoded_text,
                    "file_size": len(file.body),
                    "filename": file.name,
                    "model_used": request.form.get("model", "mock-model"),
                    "mimetype": file.type,
                    "usage": self._audio_usage(file, decoded_text),
                }
            )

    def _setup_error_handlers(self):
        """Setup error handlers."""

        @self.app.exception(Exception)
        async def generic_error_handler(_request: Request, exception: Exception):
            logger.error("Unhandled exception: %s", exception)
            return response.json(
                {
                    "error": {
                        "message": "Internal server error",
                        "type": type(exception).__name__,
                        "error": str(exception),
                    }
                },
                status=500,
            )

        @self.app.exception(NotFound)
        async def not_found_handler(_request: Request, _exception):
            return response.json(
                {
                    "error": {
                        "message": "Not Found",
                        "type": "not_found_error",
                        "code": "not_found",
                    }
                },
                status=404,
            )

    def run(self, *, access_log: bool = True) -> None:
        """
        Start the mock server with configured settings.

        Runs the Sanic application in single-process mode with access logging and
        the Sanic startup MOTD enabled by default. Pass ``access_log=False`` in
        shared-TTY test runners to disable access logs and the MOTD, and to use
        plain (non-ANSI) Sanic formatters so start/stop logs do not overwrite
        the terminal.

        :param access_log: Whether to enable Sanic per-request access logging and
            the startup MOTD banner. When false, start/stop logs still print but
            without ANSI cursor controls.
        """
        # Reconfigure after Sanic.__init__ so quiet mode applies before serve.
        _configure_sanic_logging(access_log=access_log)
        self.app.run(
            host=self.config.host,
            port=self.config.port,
            debug=False,
            single_process=True,
            access_log=access_log,
            motd=access_log,
            register_sys_signals=True,
        )