cleveragents-core/robot/helper_lsp_stub.py

"""Helper script for Robot Framework LSP stub smoke tests.

Usage:
    python helper_lsp_stub.py <command>

Commands:
    initialize       Send initialize request and verify response
    shutdown         Full shutdown handshake
    unsupported      Test unsupported method error
    startup-pid      Verify startup PID reporting
    full-lifecycle   Complete initialize/shutdown/exit lifecycle
"""

from __future__ import annotations

import io
import json
import sys
from pathlib import Path

# Ensure src is on the path.  Robot helpers are invoked as standalone
# scripts via ``Run Process`` so the project package is not always
# importable.  This sys.path manipulation is required for that use
# case.  If the project is installed in editable mode this insert is
# harmless (the installed entry-point takes precedence).
src_dir = str(Path(__file__).resolve().parents[1] / "src")
if src_dir not in sys.path:
    sys.path.insert(0, src_dir)

from cleveragents.lsp.server import LspServer  # noqa: E402

# Import the shared response-parsing utility from the test mocks
# package so that parsing logic is defined in a single place.
# Same sys.path rationale as above — needed for standalone execution.
features_dir = str(Path(__file__).resolve().parents[1])
if features_dir not in sys.path:
    sys.path.insert(0, features_dir)

from features.mocks.lsp_transport_mock import parse_lsp_responses  # noqa: E402


def _encode_message(msg: dict) -> bytes:
    """Encode a JSON-RPC message with Content-Length framing."""
    payload = json.dumps(msg).encode("utf-8")
    header = f"Content-Length: {len(payload)}\r\n\r\n".encode()
    return header + payload


def _run_server(messages: list[dict]) -> tuple[int, list[dict]]:
    """Run LSP server with the given messages and return (exit_code, responses)."""
    raw = b"".join(_encode_message(msg) for msg in messages)
    input_stream = io.BytesIO(raw)
    output_stream = io.BytesIO()
    server = LspServer(input_stream=input_stream, output_stream=output_stream)
    exit_code = server.run()
    output_stream.seek(0)
    responses = parse_lsp_responses(output_stream.read())
    return exit_code, responses


def cmd_initialize() -> None:
    """Test initialize handshake."""
    _, responses = _run_server(
        [
            {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {"capabilities": {}},
            },
            {"jsonrpc": "2.0", "method": "exit"},
        ]
    )
    assert len(responses) >= 1, f"Expected at least 1 response, got {len(responses)}"
    result = responses[0].get("result", {})
    server_info = result.get("serverInfo", {})
    assert server_info.get("name") == "cleveragents-lsp-stub", (
        f"Unexpected serverInfo: {server_info}"
    )
    assert "capabilities" in result, "Missing capabilities"
    print("lsp-initialize-ok")


def cmd_shutdown() -> None:
    """Test shutdown handshake."""
    exit_code, responses = _run_server(
        [
            {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {"capabilities": {}},
            },
            {"jsonrpc": "2.0", "id": 2, "method": "shutdown"},
            {"jsonrpc": "2.0", "method": "exit"},
        ]
    )
    assert exit_code == 0, f"Expected exit code 0, got {exit_code}"
    shutdown_resp = None
    for r in responses:
        if r.get("id") == 2:
            shutdown_resp = r
            break
    assert shutdown_resp is not None, "No shutdown response"
    assert shutdown_resp.get("result") is None, "Shutdown result should be null"
    print("lsp-shutdown-ok")


def cmd_unsupported() -> None:
    """Test unsupported method error after initialize."""
    _, responses = _run_server(
        [
            {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {"capabilities": {}},
            },
            {
                "jsonrpc": "2.0",
                "id": 2,
                "method": "textDocument/completion",
                "params": {},
            },
            {"jsonrpc": "2.0", "method": "exit"},
        ]
    )
    assert len(responses) >= 2, f"Expected at least 2 responses, got {len(responses)}"
    # Find the completion response (id=2)
    completion_resp = None
    for r in responses:
        if r.get("id") == 2:
            completion_resp = r
            break
    assert completion_resp is not None, "No response for textDocument/completion"
    error = completion_resp.get("error", {})
    assert error.get("code") == -32601, f"Expected -32601, got {error.get('code')}"
    assert "Not implemented" in error.get("message", ""), (
        f"Unexpected error message: {error.get('message')}"
    )
    print("lsp-unsupported-ok")


def cmd_startup_pid() -> None:
    """Test that server starts and processes messages (PID reported via logging).

    Validates PID plausibility (> 0) by capturing the structlog
    ``lsp.server.starting`` event emitted during ``run()``.
    """
    import structlog

    exit_code: int
    responses: list[dict]
    messages = [
        {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "initialize",
            "params": {"capabilities": {}},
        },
        {"jsonrpc": "2.0", "method": "exit"},
    ]
    raw = b"".join(_encode_message(msg) for msg in messages)
    input_stream = io.BytesIO(raw)
    output_stream = io.BytesIO()
    server = LspServer(input_stream=input_stream, output_stream=output_stream)
    with structlog.testing.capture_logs() as captured:
        exit_code = server.run()
    output_stream.seek(0)
    responses = parse_lsp_responses(output_stream.read())

    assert len(responses) >= 1, "Server should produce responses"
    # exit_code is 1 because no shutdown was sent
    assert exit_code == 1, f"Expected exit code 1 (no shutdown), got {exit_code}"

    # Validate PID plausibility: the server must log its PID as a
    # positive integer in the ``lsp.server.starting`` event.
    pid_entries = [
        e for e in captured if e.get("event") == "lsp.server.starting" and "pid" in e
    ]
    assert pid_entries, (
        f"No 'lsp.server.starting' log with PID found in captured logs: {captured}"
    )
    logged_pid = pid_entries[0]["pid"]
    assert isinstance(logged_pid, int), f"PID should be int, got {type(logged_pid)}"
    assert logged_pid > 0, f"PID should be > 0, got {logged_pid}"
    print("lsp-startup-pid-ok")


def _find_response_by_id(responses: list[dict], req_id: int) -> dict:
    """Find a response by its JSON-RPC id."""
    for r in responses:
        if r.get("id") == req_id:
            return r
    raise AssertionError(f"No response with id={req_id} in {responses}")


def cmd_full_lifecycle() -> None:
    """Test complete lifecycle."""
    exit_code, responses = _run_server(
        [
            {
                "jsonrpc": "2.0",
                "id": 1,
                "method": "initialize",
                "params": {"capabilities": {}},
            },
            {"jsonrpc": "2.0", "id": 2, "method": "shutdown"},
            {"jsonrpc": "2.0", "method": "exit"},
        ]
    )
    assert exit_code == 0, f"Expected exit code 0, got {exit_code}"
    assert len(responses) == 2, f"Expected 2 responses, got {len(responses)}"

    # Verify initialize response (by id, not index)
    init_resp = _find_response_by_id(responses, 1)
    assert "capabilities" in init_resp.get("result", {})

    # Verify shutdown response (by id, not index)
    shutdown_resp = _find_response_by_id(responses, 2)
    assert shutdown_resp.get("result") is None

    print("lsp-full-lifecycle-ok")


COMMANDS = {
    "initialize": cmd_initialize,
    "shutdown": cmd_shutdown,
    "unsupported": cmd_unsupported,
    "startup-pid": cmd_startup_pid,
    "full-lifecycle": cmd_full_lifecycle,
}


if __name__ == "__main__":
    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
        print(f"Usage: {sys.argv[0]} <{'|'.join(COMMANDS)}>", file=sys.stderr)
        sys.exit(1)
    COMMANDS[sys.argv[1]]()