cleveragents-core/features/steps/lsp_path_containment_steps.py

"""Step definitions for lsp_path_containment.feature.

Tests workspace path containment in LspRuntime._read_file to prevent
path traversal attacks.  Uses the ``lspc`` step prefix to avoid
Behave AmbiguousStep errors.
"""

from __future__ import annotations

import os
import tempfile
from unittest.mock import MagicMock

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

from cleveragents.lsp.errors import LspError
from cleveragents.lsp.lifecycle import LspLifecycleManager
from cleveragents.lsp.runtime import LspRuntime

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_mock_client() -> MagicMock:
    """Create a mock LSP client with the methods runtime calls."""
    client = MagicMock(name="mock_lsp_client")
    client.did_open = MagicMock()
    client.did_close = MagicMock()
    client.get_diagnostics = MagicMock(return_value=[])
    client.get_completions = MagicMock(return_value=[])
    client.get_hover = MagicMock(return_value=None)
    client.get_definitions = MagicMock(return_value=[])
    return client


# ---------------------------------------------------------------------------
# Given steps
# ---------------------------------------------------------------------------


@given("lspc I have a temp workspace directory")
def step_lspc_create_workspace(context: Context) -> None:
    workspace = tempfile.mkdtemp(prefix="lspc_workspace_")
    context.lspc_workspace = workspace
    context.lspc_error = None

    def cleanup() -> None:
        import shutil

        if os.path.exists(workspace):
            shutil.rmtree(workspace, ignore_errors=True)

    context.add_cleanup(cleanup)


@given('lspc I have a file inside the workspace with content "{content}"')
def step_lspc_create_inside_file(context: Context, content: str) -> None:
    fd, path = tempfile.mkstemp(
        suffix=".py",
        dir=context.lspc_workspace,
        prefix="inside_",
    )
    with os.fdopen(fd, "w") as f:
        f.write(content)
    context.lspc_inside_file = path

    def cleanup() -> None:
        if os.path.exists(path):
            os.unlink(path)

    context.add_cleanup(cleanup)


@given("lspc I have a file outside the workspace")
def step_lspc_create_outside_file(context: Context) -> None:
    fd, path = tempfile.mkstemp(suffix=".py", prefix="outside_")
    with os.fdopen(fd, "w") as f:
        f.write("outside content")
    context.lspc_outside_file = path
    context.lspc_error = None

    def cleanup() -> None:
        if os.path.exists(path):
            os.unlink(path)

    context.add_cleanup(cleanup)


@given('lspc I create an LspRuntime with a healthy mock server "{name}" and workspace')
def step_lspc_create_runtime_with_workspace(context: Context, name: str) -> None:
    mock_client = _make_mock_client()

    mock_lifecycle = MagicMock(spec=LspLifecycleManager)
    mock_lifecycle.health_check = MagicMock(return_value=True)
    mock_lifecycle.get_client = MagicMock(return_value=mock_client)
    mock_lifecycle.start_server = MagicMock()

    runtime = LspRuntime(lifecycle_manager=mock_lifecycle)
    # Register the workspace path by calling start_server
    # We need to mock the registry lookup too
    from cleveragents.lsp.models import LspServerConfig
    from cleveragents.lsp.registry import LspRegistry

    registry = LspRegistry()
    config = LspServerConfig(name=name, command="echo", languages=["python"])
    registry.register(config)

    runtime = LspRuntime(registry=registry, lifecycle_manager=mock_lifecycle)
    runtime.start_server(name, context.lspc_workspace)

    context.lspc_runtime = runtime
    context.lspc_mock_client = mock_client
    context.lspc_error = None


@given(
    'lspc I create an LspRuntime with a healthy mock server "{name}" without workspace'
)
def step_lspc_create_runtime_without_workspace(context: Context, name: str) -> None:
    mock_client = _make_mock_client()

    mock_lifecycle = MagicMock(spec=LspLifecycleManager)
    mock_lifecycle.health_check = MagicMock(return_value=True)
    mock_lifecycle.get_client = MagicMock(return_value=mock_client)

    runtime = LspRuntime(lifecycle_manager=mock_lifecycle)
    # Do NOT call start_server — no workspace path registered

    context.lspc_runtime = runtime
    context.lspc_mock_client = mock_client
    context.lspc_error = None


# ---------------------------------------------------------------------------
# When steps
# ---------------------------------------------------------------------------


@when("lspc I call read_file with the workspace path")
def step_lspc_read_file_with_workspace(context: Context) -> None:
    # Determine which file to use: inside or outside
    file_path = getattr(context, "lspc_inside_file", None) or getattr(
        context, "lspc_outside_file", None
    )
    try:
        context.lspc_file_content = LspRuntime._read_file(
            file_path, context.lspc_workspace
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_file_content = None


@when("lspc I call read_file with a traversal path and the workspace path")
def step_lspc_read_file_traversal(context: Context) -> None:
    # Build a traversal path: workspace/subdir/../../outside_file
    outside_file = context.lspc_outside_file
    workspace = context.lspc_workspace
    # Construct a path that starts inside the workspace but traverses out
    traversal_path = os.path.join(
        workspace, "subdir", "..", "..", outside_file.lstrip("/")
    )
    try:
        context.lspc_file_content = LspRuntime._read_file(traversal_path, workspace)
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_file_content = None


@when("lspc I call read_file without a workspace path")
def step_lspc_read_file_no_workspace(context: Context) -> None:
    file_path = context.lspc_outside_file
    try:
        context.lspc_file_content = LspRuntime._read_file(file_path)
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_file_content = None


@when('lspc I try to get diagnostics for "{name}" on the outside file')
def step_lspc_get_diagnostics_outside(context: Context, name: str) -> None:
    try:
        context.lspc_result = context.lspc_runtime.get_diagnostics(
            name, context.lspc_outside_file
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_result = None


@when('lspc I get diagnostics for "{name}" on the inside file')
def step_lspc_get_diagnostics_inside(context: Context, name: str) -> None:
    try:
        context.lspc_result = context.lspc_runtime.get_diagnostics(
            name, context.lspc_inside_file
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_result = None


@when('lspc I get diagnostics for "{name}" on the outside file')
def step_lspc_get_diagnostics_outside_no_ws(context: Context, name: str) -> None:
    try:
        context.lspc_result = context.lspc_runtime.get_diagnostics(
            name, context.lspc_outside_file
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_result = None


@when(
    'lspc I try to get completions for "{name}" on the outside file'
    " at line {line:d} column {col:d}"
)
def step_lspc_get_completions_outside(
    context: Context, name: str, line: int, col: int
) -> None:
    try:
        context.lspc_result = context.lspc_runtime.get_completions(
            name, context.lspc_outside_file, line, col
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_result = None


@when(
    'lspc I try to get hover for "{name}" on the outside file'
    " at line {line:d} column {col:d}"
)
def step_lspc_get_hover_outside(
    context: Context, name: str, line: int, col: int
) -> None:
    try:
        context.lspc_result = context.lspc_runtime.get_hover(
            name, context.lspc_outside_file, line, col
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_result = None


@when(
    'lspc I try to get definitions for "{name}" on the outside file'
    " at line {line:d} column {col:d}"
)
def step_lspc_get_definitions_outside(
    context: Context, name: str, line: int, col: int
) -> None:
    try:
        context.lspc_result = context.lspc_runtime.get_definitions(
            name, context.lspc_outside_file, line, col
        )
        context.lspc_error = None
    except Exception as exc:
        context.lspc_error = exc
        context.lspc_result = None


# ---------------------------------------------------------------------------
# Then steps
# ---------------------------------------------------------------------------


@then('lspc the file content should be "{expected}"')
def step_lspc_file_content(context: Context, expected: str) -> None:
    assert context.lspc_error is None, f"Expected no error, got {context.lspc_error}"
    assert context.lspc_file_content == expected, (
        f"Expected '{expected}', got '{context.lspc_file_content}'"
    )


@then("lspc no error should be raised")
def step_lspc_no_error(context: Context) -> None:
    assert context.lspc_error is None, (
        f"Expected no error, got {type(context.lspc_error).__name__}: "
        f"{context.lspc_error}"
    )


@then('lspc an LspError should be raised with message containing "{msg}"')
def step_lspc_lsp_error_msg(context: Context, msg: str) -> None:
    assert context.lspc_error is not None, "Expected an LspError but no error occurred"
    assert isinstance(context.lspc_error, LspError), (
        f"Expected LspError, got {type(context.lspc_error).__name__}: "
        f"{context.lspc_error}"
    )
    assert msg in str(context.lspc_error), (
        f"Expected '{msg}' in error message, got: {context.lspc_error}"
    )


@then("lspc diagnostics should be returned as a list")
def step_lspc_diagnostics_is_list(context: Context) -> None:
    assert context.lspc_error is None, f"Expected no error, got {context.lspc_error}"
    assert isinstance(context.lspc_result, list), (
        f"Expected list, got {type(context.lspc_result)}"
    )