forked from cleveragents/cleveragents-core
HAL9000
4f23ece138
Fix CI lint failure by running ruff format on the step definitions file that was missing auto-formatting before the PR was submitted.
311 lines
10 KiB
Python
311 lines
10 KiB
Python
"""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)}"
|
|
)
|