cleveragents-core/robot/helper_lsp_functional_runtime.py

"""Robot Framework helper for LSP functional runtime integration tests."""

from __future__ import annotations

import importlib
import sys
from pathlib import Path

# Ensure local source tree is importable
_SRC = str(Path(__file__).resolve().parents[1] / "src")
if _SRC not in sys.path:
    sys.path.insert(0, _SRC)

import cleveragents  # noqa: E402

importlib.reload(cleveragents)

from cleveragents.lsp.discovery import LanguageDiscovery  # noqa: E402
from cleveragents.lsp.errors import LspServerNotFoundError  # noqa: E402
from cleveragents.lsp.lifecycle import LspLifecycleManager  # noqa: E402
from cleveragents.lsp.models import LspCapability, LspServerConfig  # noqa: E402
from cleveragents.lsp.runtime import LspRuntime  # noqa: E402
from cleveragents.lsp.tool_adapter import LspToolAdapter  # noqa: E402
from cleveragents.lsp.transport import StdioTransport  # noqa: E402


def _run_transport_lifecycle() -> None:
    """Test transport create/alive check."""
    t = StdioTransport(command="cat")
    assert not t.is_alive
    t.start()
    assert t.is_alive
    t.stop()
    assert not t.is_alive
    print("transport-lifecycle-ok")


def _run_transport_empty_command() -> None:
    """Test transport rejects empty command."""
    try:
        StdioTransport(command="")
        print("FAIL: expected ValueError")
        sys.exit(1)
    except ValueError:
        pass
    print("transport-empty-command-ok")


def _run_lifecycle_manager_empty() -> None:
    """Test lifecycle manager starts empty."""
    mgr = LspLifecycleManager()
    assert len(mgr.list_running()) == 0
    assert not mgr.health_check("local/test")
    try:
        mgr.stop_server("local/test")
        print("FAIL: expected LspServerNotFoundError")
        sys.exit(1)
    except LspServerNotFoundError:
        pass
    print("lifecycle-manager-empty-ok")


def _run_runtime_validation() -> None:
    """Test runtime input validation."""
    rt = LspRuntime()
    errors = 0

    try:
        rt.start_server("", "/tmp")
    except ValueError:
        errors += 1

    try:
        rt.start_server("local/test", "")
    except ValueError:
        errors += 1

    try:
        rt.start_server("local/missing", "/tmp")
    except LspServerNotFoundError:
        errors += 1

    assert errors == 3, f"Expected 3 validation errors, got {errors}"
    print("runtime-validation-ok")


def _run_discovery() -> None:
    """Test language discovery."""
    d = LanguageDiscovery()
    assert d.detect_file_language("test.py") == "python"
    assert d.detect_file_language("app.ts") == "typescript"
    assert d.detect_file_language("main.rs") == "rust"
    assert d.detect_file_language("unknown.xyz") == "plaintext"
    assert d.detect_file_language("Dockerfile") == "dockerfile"

    d2 = LanguageDiscovery(project_languages=["go"])
    assert d2.detect_file_language("no_ext") == "go"

    print("discovery-ok")


def _run_tool_adapter_modes() -> None:
    """Test tool adapter local vs runtime mode."""
    config = LspServerConfig(
        name="local/test",
        command="echo",
        languages=["python"],
        capabilities=[LspCapability.DIAGNOSTICS, LspCapability.COMPLETIONS],
    )

    # Local mode
    adapter_local = LspToolAdapter()
    specs_local = adapter_local.generate_tool_specs(config)
    assert len(specs_local) == 2
    assert specs_local[0]["name"] == "local/test/diagnostics"

    # Runtime mode
    adapter_rt = LspToolAdapter(runtime=LspRuntime())
    specs_rt = adapter_rt.generate_tool_specs(config)
    assert len(specs_rt) == 2
    assert specs_rt[0]["name"] == "local/test/diagnostics"

    print("tool-adapter-modes-ok")


if __name__ == "__main__":
    cmd = sys.argv[1] if len(sys.argv) > 1 else "all"
    dispatch = {
        "transport-lifecycle": _run_transport_lifecycle,
        "transport-empty-command": _run_transport_empty_command,
        "lifecycle-manager-empty": _run_lifecycle_manager_empty,
        "runtime-validation": _run_runtime_validation,
        "discovery": _run_discovery,
        "tool-adapter-modes": _run_tool_adapter_modes,
    }

    if cmd == "all":
        for fn in dispatch.values():
            fn()
    elif cmd in dispatch:
        dispatch[cmd]()
    else:
        print(f"Unknown command: {cmd}")
        sys.exit(1)