cleveragents-core/features/steps/lsp_functional_runtime_steps.py

"""Step definitions for the functional LSP runtime feature tests."""

from __future__ import annotations

import os
import tempfile
from unittest.mock import MagicMock, create_autospec

from behave import given, then, when
from behave.runner import Context

from cleveragents.lsp.client import LspClient
from cleveragents.lsp.discovery import LanguageDiscovery
from cleveragents.lsp.errors import LspNotAvailableError, LspServerNotFoundError
from cleveragents.lsp.lifecycle import LspLifecycleManager
from cleveragents.lsp.models import LspCapability, LspServerConfig
from cleveragents.lsp.runtime import LspRuntime
from cleveragents.lsp.tool_adapter import LspToolAdapter
from cleveragents.lsp.transport import StdioTransport

# ---------------------------------------------------------------------------
# StdioTransport steps
# ---------------------------------------------------------------------------


@when("I try to create a StdioTransport with empty command")
def step_transport_empty_command(context: Context) -> None:
    try:
        StdioTransport(command="")
        context.transport_error = None
    except ValueError as exc:
        context.transport_error = exc


@then("a ValueError should be raised for transport")
def step_transport_value_error(context: Context) -> None:
    assert context.transport_error is not None
    assert isinstance(context.transport_error, ValueError)


@given('I create a StdioTransport for "{cmd}"')
def step_create_transport(context: Context, cmd: str) -> None:
    context.transport = StdioTransport(command=cmd)


@then("the transport should not be alive")
def step_transport_not_alive(context: Context) -> None:
    assert not context.transport.is_alive


@when("I start the transport")
def step_start_transport(context: Context) -> None:
    context.transport.start()


@then("the transport should be alive")
def step_transport_alive(context: Context) -> None:
    assert context.transport.is_alive


@when("I stop the transport")
def step_stop_transport(context: Context) -> None:
    context.transport.stop()


# ---------------------------------------------------------------------------
# LspClient steps
# ---------------------------------------------------------------------------


@given("I create an LspClient with a mock transport")
def step_create_client_mock(context: Context) -> None:
    mock_transport = create_autospec(StdioTransport, instance=True)
    context.lsp_client = LspClient(mock_transport, server_name="test/mock")


@then("the client should not be initialized")
def step_client_not_initialized(context: Context) -> None:
    assert not context.lsp_client.is_initialized


@when("I get two request IDs from the client")
def step_get_request_ids(context: Context) -> None:
    context.id1 = context.lsp_client._next_id()
    context.id2 = context.lsp_client._next_id()


@then("the IDs should be sequential")
def step_ids_sequential(context: Context) -> None:
    assert context.id2 == context.id1 + 1


# ---------------------------------------------------------------------------
# LspLifecycleManager steps
# ---------------------------------------------------------------------------


@given("I create an LspLifecycleManager")
def step_create_lifecycle(context: Context) -> None:
    context.lifecycle_mgr = LspLifecycleManager()


@then("the manager should have no running servers")
def step_lifecycle_empty(context: Context) -> None:
    assert len(context.lifecycle_mgr.list_running()) == 0


@when('I try to stop server "{name}" on the manager')
def step_lifecycle_stop_unknown(context: Context, name: str) -> None:
    try:
        context.lifecycle_mgr.stop_server(name)
        context.lifecycle_error = None
    except LspServerNotFoundError as exc:
        context.lifecycle_error = exc


@then("an LspServerNotFoundError should be raised for lifecycle")
def step_lifecycle_not_found(context: Context) -> None:
    assert context.lifecycle_error is not None
    assert isinstance(context.lifecycle_error, LspServerNotFoundError)


@then('health check for "{name}" should return false')
def step_health_false(context: Context, name: str) -> None:
    assert not context.lifecycle_mgr.health_check(name)


# ---------------------------------------------------------------------------
# LspRuntime steps
# ---------------------------------------------------------------------------


@given("I create a functional LspRuntime")
def step_create_runtime(context: Context) -> None:
    context.func_runtime = LspRuntime()
    context.runtime_error = None


@when('I try to start server with empty name and workspace "{ws}"')
def step_runtime_empty_name(context: Context, ws: str) -> None:
    try:
        context.func_runtime.start_server("", ws)
        context.runtime_error = None
    except ValueError as exc:
        context.runtime_error = exc


@when('I try to start server "{name}" with empty workspace')
def step_runtime_empty_workspace(context: Context, name: str) -> None:
    try:
        context.func_runtime.start_server(name, "")
        context.runtime_error = None
    except ValueError as exc:
        context.runtime_error = exc


@when('I try to start unregistered server "{name}"')
def step_runtime_unregistered(context: Context, name: str) -> None:
    try:
        context.func_runtime.start_server(name, "/tmp")
        context.runtime_error = None
    except LspServerNotFoundError as exc:
        context.runtime_error = exc


@then("a ValueError should be raised for runtime")
def step_runtime_value_error(context: Context) -> None:
    assert context.runtime_error is not None
    assert isinstance(context.runtime_error, ValueError)


@then("an LspServerNotFoundError should be raised for runtime")
def step_runtime_not_found(context: Context) -> None:
    assert context.runtime_error is not None
    assert isinstance(context.runtime_error, LspServerNotFoundError)


@when("I try to get diagnostics with empty name")
def step_runtime_diag_empty(context: Context) -> None:
    try:
        context.func_runtime.get_diagnostics("", "/tmp/test.py")
        context.runtime_error = None
    except ValueError as exc:
        context.runtime_error = exc


@when("I try to get completions with line 0")
def step_runtime_completions_line0(context: Context) -> None:
    try:
        context.func_runtime.get_completions("local/test", "/tmp/test.py", 0, 1)
        context.runtime_error = None
    except ValueError as exc:
        context.runtime_error = exc


@when("I try to get completions with column 0")
def step_runtime_completions_col0(context: Context) -> None:
    try:
        context.func_runtime.get_completions("local/test", "/tmp/test.py", 1, 0)
        context.runtime_error = None
    except ValueError as exc:
        context.runtime_error = exc


# ---------------------------------------------------------------------------
# LanguageDiscovery steps
# ---------------------------------------------------------------------------


@given("I create a LanguageDiscovery instance")
def step_create_discovery(context: Context) -> None:
    context.discovery = LanguageDiscovery()


@given('I create a LanguageDiscovery with project language "{lang}"')
def step_create_discovery_project(context: Context, lang: str) -> None:
    context.discovery = LanguageDiscovery(project_languages=[lang])


@then('language for "{filename}" should be "{expected}"')
def step_check_language(context: Context, filename: str, expected: str) -> None:
    result = context.discovery.detect_file_language(filename)
    assert result == expected, f"Expected '{expected}', got '{result}'"


@when('I detect language for "{filename}"')
def step_detect_language(context: Context, filename: str) -> None:
    context.detected_lang = context.discovery.detect_file_language(filename)


@when('I invalidate "{filename}" from discovery cache')
def step_invalidate_cache(context: Context, filename: str) -> None:
    context.discovery.invalidate(filename)


@then('the cache should not contain "{filename}"')
def step_cache_not_contains(context: Context, filename: str) -> None:
    assert filename not in context.discovery._cache


@given('I create a temp file with shebang "{shebang}"')
def step_create_shebang_file(context: Context, shebang: str) -> None:
    fd, path = tempfile.mkstemp(suffix="")
    with os.fdopen(fd, "w") as f:
        f.write(f"{shebang}\nprint('hello')\n")
    context.temp_shebang_file = path


@then('the temp file language should be "{lang}"')
def step_temp_file_lang(context: Context, lang: str) -> None:
    result = context.discovery.detect_file_language(context.temp_shebang_file)
    assert result == lang, f"Expected '{lang}', got '{result}'"
    os.unlink(context.temp_shebang_file)


# ---------------------------------------------------------------------------
# LspToolAdapter steps
# ---------------------------------------------------------------------------


@given("I create an LspToolAdapter with a functional runtime")
def step_adapter_with_runtime(context: Context) -> None:
    context.adapter_runtime = LspRuntime()
    context.tool_adapter = LspToolAdapter(runtime=context.adapter_runtime)


@given("I create an LspToolAdapter without runtime")
def step_adapter_no_runtime(context: Context) -> None:
    context.tool_adapter = LspToolAdapter()


@given("I have an LSP server config with diagnostics capability")
def step_server_config_diag(context: Context) -> None:
    context.adapter_config = LspServerConfig(
        name="local/test-server",
        command="echo",
        languages=["python"],
        capabilities=[LspCapability.DIAGNOSTICS],
    )


@when("I generate functional tool specs from the adapter")
def step_generate_specs(context: Context) -> None:
    context.generated_specs = context.tool_adapter.generate_tool_specs(
        context.adapter_config,
    )


@then("the tool specs should have runtime handlers")
def step_specs_runtime_handlers(context: Context) -> None:
    assert len(context.generated_specs) > 0
    spec = context.generated_specs[0]
    assert spec["name"] == "local/test-server/diagnostics"
    # Runtime handler should not raise LspNotAvailableError
    # (it would raise LspServerNotFoundError instead since server not started)
    handler = spec["handler"]
    try:
        handler(file_path="/tmp/test.py")
    except LspNotAvailableError as exc:
        raise AssertionError(
            "Expected runtime handler, got local-mode handler"
        ) from exc
    except Exception:
        pass  # Expected: server not started


@then("the tool specs should have local-mode handlers")
def step_specs_local_handlers(context: Context) -> None:
    assert len(context.generated_specs) > 0
    spec = context.generated_specs[0]
    handler = spec["handler"]
    try:
        handler(file_path="/tmp/test.py")
        raise AssertionError("Expected LspNotAvailableError")
    except LspNotAvailableError:
        pass  # Expected


# ---------------------------------------------------------------------------
# Binding activation steps
# ---------------------------------------------------------------------------


@when("I activate bindings with empty list")
def step_activate_empty(context: Context) -> None:
    context.activated = context.func_runtime.activate_bindings([], "/tmp")


@then("the activated servers list should be empty")
def step_activated_empty(context: Context) -> None:
    assert context.activated == []


@when("I activate bindings with unregistered server binding")
def step_activate_unregistered(context: Context) -> None:
    mock_binding = MagicMock()
    mock_binding.lsp_server_name = "local/nonexistent"
    mock_binding.node_name = "test_node"
    context.activated = context.func_runtime.activate_bindings(
        [mock_binding],
        "/tmp",
    )