cleveragents-core/features/mocks/lsp_transport_mock.py

"""Mock stdin/stdout transport for LSP server testing.

Provides :class:`MockLspTransport` which simulates the
Content-Length framed JSON-RPC transport used by the LSP server
without requiring a real subprocess.

The :func:`parse_lsp_responses` utility is shared with the Robot
Framework helper to avoid duplicating response-parsing logic.

All mocks reside in ``features/mocks/`` per project conventions.
"""

from __future__ import annotations

import io
import json
from typing import Any


def parse_lsp_responses(raw: bytes) -> list[dict[str, Any]]:
    """Parse Content-Length framed JSON-RPC responses from raw bytes.

    Shared utility used by :class:`MockLspTransport` and the Robot
    Framework helper to avoid duplicating the parsing logic.

    Args:
        raw: The raw bytes containing one or more Content-Length
            framed JSON-RPC response messages.

    Returns:
        List of parsed JSON-RPC response dicts.
    """
    responses: list[dict[str, Any]] = []
    offset = 0
    while offset < len(raw):
        header_end = raw.find(b"\r\n\r\n", offset)
        if header_end == -1:
            break
        header_section = raw[offset:header_end].decode("utf-8", errors="replace")
        content_length: int | None = None
        for line in header_section.split("\r\n"):
            if line.lower().startswith("content-length:"):
                content_length = int(line.split(":", 1)[1].strip())
        if content_length is None:
            break
        body_start = header_end + 4
        body_end = body_start + content_length
        body = raw[body_start:body_end]
        responses.append(json.loads(body))
        offset = body_end
    return responses


class MockLspTransport:
    """Simulated LSP stdin/stdout binary streams.

    Encodes JSON-RPC messages with Content-Length framing into a
    binary input buffer and captures binary output for assertion.

    Usage::

        transport = MockLspTransport()
        transport.send_request("initialize", {"capabilities": {}}, id=1)
        server = LspServer(
            input_stream=transport.input_stream,
            output_stream=transport.output_stream,
        )
        server.run()
        responses = transport.read_responses()
    """

    def __init__(self) -> None:
        self._input_buf = io.BytesIO()
        self._output_buf = io.BytesIO()

    def send_request(
        self,
        method: str,
        params: dict[str, Any] | None = None,
        request_id: int | str | None = 1,
    ) -> None:
        """Append a JSON-RPC request to the input buffer.

        Args:
            method: The JSON-RPC method name.
            params: Optional parameters dict.
            request_id: The request id (``None`` for notifications).
        """
        msg: dict[str, Any] = {"jsonrpc": "2.0", "method": method}
        if request_id is not None:
            msg["id"] = request_id
        if params is not None:
            msg["params"] = params
        self._append_message(msg)

    def send_notification(
        self,
        method: str,
        params: dict[str, Any] | None = None,
    ) -> None:
        """Append a JSON-RPC notification (no id) to the input buffer.

        Args:
            method: The JSON-RPC method name.
            params: Optional parameters dict.
        """
        self.send_request(method, params, request_id=None)

    def send_raw(self, raw_bytes: bytes) -> None:
        """Append raw bytes to the input buffer.

        Useful for testing malformed messages.

        Args:
            raw_bytes: The raw bytes to append.
        """
        pos = self._input_buf.tell()
        self._input_buf.seek(0, 2)
        self._input_buf.write(raw_bytes)
        self._input_buf.seek(pos)

    def _append_message(self, msg: dict[str, Any]) -> None:
        """Encode and append a Content-Length framed message."""
        payload = json.dumps(msg).encode("utf-8")
        header = f"Content-Length: {len(payload)}\r\n\r\n".encode()
        pos = self._input_buf.tell()
        self._input_buf.seek(0, 2)
        self._input_buf.write(header + payload)
        self._input_buf.seek(pos)

    @property
    def input_stream(self) -> io.BytesIO:
        """Return the input stream positioned at the start."""
        self._input_buf.seek(0)
        return self._input_buf

    @property
    def output_stream(self) -> io.BytesIO:
        """Return the output stream for writing."""
        return self._output_buf

    def read_responses(self) -> list[dict[str, Any]]:
        """Parse all JSON-RPC responses from the output buffer.

        Returns:
            List of parsed JSON-RPC response dicts.
        """
        self._output_buf.seek(0)
        raw = self._output_buf.read()
        if not raw:
            return []
        return parse_lsp_responses(raw)


__all__ = [
    "MockLspTransport",
    "parse_lsp_responses",
]