freemo
31472b5413
CI / benchmark-publish (pull_request) Has been skipped
CI / build (pull_request) Successful in 21s
CI / helm (pull_request) Successful in 22s
CI / lint (pull_request) Successful in 3m20s
CI / quality (pull_request) Successful in 3m43s
CI / typecheck (pull_request) Successful in 3m58s
CI / security (pull_request) Successful in 4m8s
CI / integration_tests (pull_request) Successful in 9m33s
CI / unit_tests (pull_request) Successful in 10m12s
CI / docker (pull_request) Successful in 1m28s
CI / coverage (pull_request) Successful in 12m18s
CI / e2e_tests (pull_request) Successful in 19m51s
CI / status-check (pull_request) Successful in 1s
CI / build (push) Successful in 15s
CI / helm (push) Successful in 22s
CI / lint (push) Successful in 3m18s
CI / quality (push) Successful in 3m41s
CI / typecheck (push) Successful in 3m57s
CI / benchmark-regression (push) Has been skipped
CI / security (push) Successful in 4m10s
CI / unit_tests (push) Failing after 6m58s
CI / docker (push) Has been skipped
CI / integration_tests (push) Successful in 9m15s
CI / coverage (push) Successful in 12m26s
CI / e2e_tests (push) Successful in 23m11s
CI / status-check (push) Failing after 1s
CI / benchmark-publish (push) Successful in 28m31s
CI / benchmark-regression (pull_request) Successful in 54m53s
Add Behave feature/step pairs that exercise uncovered branches across handlers, LSP, CLI, and service layers to reach the coverage gate. ISSUES CLOSED: #1232
640 lines
23 KiB
Python
640 lines
23 KiB
Python
"""Step definitions for lsp_client_coverage.feature.
|
|
|
|
Covers every uncovered code-path in ``src/cleveragents/lsp/client.py``.
|
|
Uses the ``lccov`` prefix on every step to avoid Behave AmbiguousStep errors.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
from unittest.mock import MagicMock, PropertyMock, create_autospec
|
|
|
|
from behave import given, then, when
|
|
|
|
from cleveragents.lsp.client import LspClient
|
|
from cleveragents.lsp.errors import LspError
|
|
from cleveragents.lsp.transport import StdioTransport
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_mock_transport() -> MagicMock:
|
|
"""Create an ``autospec``-ed mock of :class:`StdioTransport`."""
|
|
mock = create_autospec(StdioTransport, instance=True)
|
|
# Default: transport is alive
|
|
type(mock).is_alive = PropertyMock(return_value=True)
|
|
return mock
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Given steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@given("lccov I create an LspClient with a mock transport")
|
|
def step_create_client(context: Any) -> None:
|
|
context.lccov_transport = _make_mock_transport()
|
|
context.lccov_client = LspClient(context.lccov_transport, server_name="test/lccov")
|
|
# Track sent messages for assertions
|
|
context.lccov_sent_messages = []
|
|
|
|
def _capture_send(msg: dict) -> None:
|
|
context.lccov_sent_messages.append(msg)
|
|
|
|
context.lccov_transport.send_message = MagicMock(side_effect=_capture_send)
|
|
|
|
|
|
@given("lccov the transport echoes back a successful response")
|
|
def step_transport_echo_success(context: Any) -> None:
|
|
"""Configure read_message to return a response matching the current request id.
|
|
|
|
By the time read_message is called, _next_id() has already incremented
|
|
_request_id, so _request_id holds the id used by the in-flight request.
|
|
"""
|
|
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": {"status": "ok"},
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport returns a JSON-RPC error response")
|
|
def step_transport_rpc_error(context: Any) -> None:
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"error": {
|
|
"code": -32600,
|
|
"message": "Invalid Request",
|
|
"data": "bad stuff",
|
|
},
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport returns None and is not alive")
|
|
def step_transport_dead(context: Any) -> None:
|
|
context.lccov_transport.read_message = MagicMock(return_value=None)
|
|
type(context.lccov_transport).is_alive = PropertyMock(return_value=False)
|
|
|
|
|
|
@given("lccov the transport returns None then a valid response while alive")
|
|
def step_transport_retry_then_success(context: Any) -> None:
|
|
"""First call returns None (alive so retry), second returns valid response."""
|
|
call_count = {"n": 0}
|
|
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
return None # First read: no data, but transport is alive
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": {"status": "ok"},
|
|
}
|
|
|
|
type(context.lccov_transport).is_alive = PropertyMock(return_value=True)
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport always returns None but stays alive")
|
|
def step_transport_always_none_alive(context: Any) -> None:
|
|
context.lccov_transport.read_message = MagicMock(return_value=None)
|
|
type(context.lccov_transport).is_alive = PropertyMock(return_value=True)
|
|
|
|
|
|
@given("lccov the transport returns a notification then the response")
|
|
def step_transport_notification_then_response(context: Any) -> None:
|
|
call_count = {"n": 0}
|
|
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
# A notification (no matching id)
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"method": "window/logMessage",
|
|
"params": {"type": 3, "message": "hello"},
|
|
}
|
|
# The actual response
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": {"status": "ok"},
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport echoes back an initialize result with capabilities")
|
|
def step_transport_initialize_result(context: Any) -> None:
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": {
|
|
"capabilities": {
|
|
"textDocumentSync": 1,
|
|
"completionProvider": {"triggerCharacters": ["."]},
|
|
},
|
|
},
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the client is already marked as initialized")
|
|
def step_client_force_initialized(context: Any) -> None:
|
|
context.lccov_client._initialized = True
|
|
|
|
|
|
@given("lccov the transport echoes back a shutdown response")
|
|
def step_transport_shutdown_response(context: Any) -> None:
|
|
"""Shutdown returns a null result; exit is a notification (no read)."""
|
|
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {"jsonrpc": "2.0", "id": req_id, "result": None}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport will raise LspError on shutdown request")
|
|
def step_transport_shutdown_lsp_error(context: Any) -> None:
|
|
"""Make read_message return an error response for the shutdown request."""
|
|
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"error": {
|
|
"code": -32603,
|
|
"message": "Internal error",
|
|
},
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given(
|
|
"lccov the transport echoes back a shutdown response but raises BrokenPipeError on exit"
|
|
)
|
|
def step_transport_broken_pipe_on_exit(context: Any) -> None:
|
|
"""read_message returns valid shutdown response, send_message raises on 2nd call."""
|
|
call_count = {"n": 0}
|
|
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {"jsonrpc": "2.0", "id": req_id, "result": None}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
original_messages: list[dict] = context.lccov_sent_messages
|
|
|
|
def _send_side_effect(msg: dict) -> None:
|
|
call_count["n"] += 1
|
|
original_messages.append(msg)
|
|
# First send = shutdown request, second = exit notification
|
|
if call_count["n"] >= 2:
|
|
raise BrokenPipeError("pipe broken")
|
|
|
|
context.lccov_transport.send_message = MagicMock(side_effect=_send_side_effect)
|
|
|
|
|
|
@given('lccov there are cached diagnostics for "{uri}"')
|
|
def step_cache_diagnostics(context: Any, uri: str) -> None:
|
|
context.lccov_client._diagnostics[uri] = [
|
|
{"range": {}, "message": "old warning", "severity": 2},
|
|
]
|
|
|
|
|
|
@given("lccov the transport has queued diagnostic notifications for drain")
|
|
def step_transport_queued_diagnostics(context: Any) -> None:
|
|
"""read_message returns a publishDiagnostics notification, then None."""
|
|
call_count = {"n": 0}
|
|
|
|
def _read_side_effect(timeout: float = 0.05) -> dict | None:
|
|
call_count["n"] += 1
|
|
if call_count["n"] == 1:
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"method": "textDocument/publishDiagnostics",
|
|
"params": {
|
|
"uri": "file:///test.py",
|
|
"diagnostics": [
|
|
{"range": {}, "message": "err", "severity": 1},
|
|
],
|
|
},
|
|
}
|
|
return None
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport returns None immediately for drain")
|
|
def step_transport_drain_none(context: Any) -> None:
|
|
context.lccov_transport.read_message = MagicMock(return_value=None)
|
|
|
|
|
|
@given("lccov the transport returns a CompletionList response")
|
|
def step_transport_completion_list(context: Any) -> None:
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": {
|
|
"isIncomplete": False,
|
|
"items": [
|
|
{"label": "foo", "kind": 6},
|
|
{"label": "bar", "kind": 6},
|
|
],
|
|
},
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport returns a plain list completion response")
|
|
def step_transport_completion_plain_list(context: Any) -> None:
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": [{"label": "baz", "kind": 6}],
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
@given("lccov the transport returns a scalar completion response")
|
|
def step_transport_completion_scalar(context: Any) -> None:
|
|
def _read_side_effect(timeout: float = 30.0) -> dict | None:
|
|
req_id = context.lccov_client._request_id
|
|
return {
|
|
"jsonrpc": "2.0",
|
|
"id": req_id,
|
|
"result": "unexpected_string",
|
|
}
|
|
|
|
context.lccov_transport.read_message = MagicMock(side_effect=_read_side_effect)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# When steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@when('lccov I call _send_request with method "{method}" and params')
|
|
def step_send_request_with_params(context: Any, method: str) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client._send_request(
|
|
method, {"key": "value"}, timeout=2.0
|
|
)
|
|
except Exception as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when('lccov I call _send_request with method "{method}" and no params')
|
|
def step_send_request_no_params(context: Any, method: str) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client._send_request(method, timeout=2.0)
|
|
except Exception as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when("lccov I call _send_request expecting an error")
|
|
def step_send_request_expect_error(context: Any) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client._send_request(
|
|
"test/fail", timeout=2.0
|
|
)
|
|
except LspError as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when("lccov I call _send_request with a very short timeout expecting error")
|
|
def step_send_request_short_timeout(context: Any) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client._send_request(
|
|
"test/timeout", timeout=0.01
|
|
)
|
|
except LspError as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when('lccov I call _send_notification with method "{method}" and params')
|
|
def step_send_notification_with_params(context: Any, method: str) -> None:
|
|
context.lccov_client._send_notification(method, {"data": "test"})
|
|
|
|
|
|
@when('lccov I call _send_notification with method "{method}" and no params')
|
|
def step_send_notification_no_params(context: Any, method: str) -> None:
|
|
context.lccov_client._send_notification(method)
|
|
|
|
|
|
@when('lccov I handle a publishDiagnostics notification for "{uri}"')
|
|
def step_handle_publish_diagnostics(context: Any, uri: str) -> None:
|
|
context.lccov_client._handle_notification(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": "textDocument/publishDiagnostics",
|
|
"params": {
|
|
"uri": uri,
|
|
"diagnostics": [
|
|
{"range": {}, "message": "err1", "severity": 1},
|
|
{"range": {}, "message": "warn1", "severity": 2},
|
|
],
|
|
},
|
|
}
|
|
)
|
|
|
|
|
|
@when("lccov I handle a window/logMessage notification")
|
|
def step_handle_log_notification(context: Any) -> None:
|
|
context.lccov_client._handle_notification(
|
|
{
|
|
"jsonrpc": "2.0",
|
|
"method": "window/logMessage",
|
|
"params": {"type": 3, "message": "info msg"},
|
|
}
|
|
)
|
|
|
|
|
|
@when('lccov I call initialize with workspace "{path}"')
|
|
def step_call_initialize(context: Any, path: str) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client.initialize(path)
|
|
except LspError as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when('lccov I call initialize with workspace "{path}" and init options')
|
|
def step_call_initialize_with_options(context: Any, path: str) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client.initialize(
|
|
path, initialization_options={"custom": True}
|
|
)
|
|
except LspError as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when("lccov I call initialize expecting an error")
|
|
def step_call_initialize_error(context: Any) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_result = context.lccov_client.initialize("/tmp/nope")
|
|
except LspError as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when("lccov I call shutdown")
|
|
def step_call_shutdown(context: Any) -> None:
|
|
context.lccov_error = None
|
|
try:
|
|
context.lccov_client.shutdown()
|
|
except Exception as exc:
|
|
context.lccov_error = exc
|
|
|
|
|
|
@when('lccov I call did_open for "{uri}" with language "{lang}"')
|
|
def step_call_did_open(context: Any, uri: str, lang: str) -> None:
|
|
context.lccov_client.did_open(uri, lang, 1, "print('hello')")
|
|
|
|
|
|
@when('lccov I call did_close for "{uri}"')
|
|
def step_call_did_close(context: Any, uri: str) -> None:
|
|
context.lccov_client.did_close(uri)
|
|
|
|
|
|
@when('lccov I call get_diagnostics for "{uri}"')
|
|
def step_call_get_diagnostics(context: Any, uri: str) -> None:
|
|
context.lccov_diag_result = context.lccov_client.get_diagnostics(uri)
|
|
|
|
|
|
@when('lccov I call get_completions for "{uri}" at line {line:d} char {char:d}')
|
|
def step_call_get_completions(context: Any, uri: str, line: int, char: int) -> None:
|
|
context.lccov_completions = context.lccov_client.get_completions(uri, line, char)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Then steps
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@then("lccov server_capabilities should be an empty dict")
|
|
def step_capabilities_empty(context: Any) -> None:
|
|
assert context.lccov_client.server_capabilities == {}
|
|
|
|
|
|
@then("lccov the request result should contain the expected data")
|
|
def step_result_expected(context: Any) -> None:
|
|
assert context.lccov_error is None, f"Unexpected error: {context.lccov_error}"
|
|
assert context.lccov_result == {"status": "ok"}
|
|
|
|
|
|
@then("lccov the sent message should include params")
|
|
def step_sent_has_params(context: Any) -> None:
|
|
# The first sent message should be the request
|
|
msg = context.lccov_sent_messages[0]
|
|
assert "params" in msg
|
|
assert msg["params"] == {"key": "value"}
|
|
|
|
|
|
@then("lccov the sent message should not include params key")
|
|
def step_sent_no_params(context: Any) -> None:
|
|
msg = context.lccov_sent_messages[0]
|
|
assert "params" not in msg
|
|
|
|
|
|
@then('lccov an LspError should be stored with message containing "{fragment}"')
|
|
def step_lsp_error_contains(context: Any, fragment: str) -> None:
|
|
assert context.lccov_error is not None, "Expected an LspError but none was raised"
|
|
assert isinstance(context.lccov_error, LspError), (
|
|
f"Expected LspError, got {type(context.lccov_error)}"
|
|
)
|
|
assert fragment in str(context.lccov_error), (
|
|
f"'{fragment}' not found in '{context.lccov_error}'"
|
|
)
|
|
|
|
|
|
@then("lccov the pending notifications should contain the queued message")
|
|
def step_pending_notifications(context: Any) -> None:
|
|
pending = context.lccov_client._pending_notifications
|
|
assert len(pending) >= 1
|
|
found = any(m.get("method") == "window/logMessage" for m in pending)
|
|
assert found, f"Expected window/logMessage in pending, got {list(pending)}"
|
|
|
|
|
|
@then("lccov the transport should have received a notification with params")
|
|
def step_notification_with_params(context: Any) -> None:
|
|
assert len(context.lccov_sent_messages) == 1
|
|
msg = context.lccov_sent_messages[0]
|
|
assert msg["method"] == "test/note"
|
|
assert "params" in msg
|
|
assert msg["params"] == {"data": "test"}
|
|
# Notifications have no "id"
|
|
assert "id" not in msg
|
|
|
|
|
|
@then("lccov the transport should have received a notification without params key")
|
|
def step_notification_no_params(context: Any) -> None:
|
|
assert len(context.lccov_sent_messages) == 1
|
|
msg = context.lccov_sent_messages[0]
|
|
assert msg["method"] == "test/note"
|
|
assert "params" not in msg
|
|
assert "id" not in msg
|
|
|
|
|
|
@then('lccov the diagnostics cache for "{uri}" should have {count:d} items')
|
|
def step_diagnostics_count(context: Any, uri: str, count: int) -> None:
|
|
diags = context.lccov_client._diagnostics.get(uri, [])
|
|
assert len(diags) == count, (
|
|
f"Expected {count} diagnostics for {uri}, got {len(diags)}"
|
|
)
|
|
|
|
|
|
@then("lccov the pending notifications deque should have {count:d} item")
|
|
def step_pending_count(context: Any, count: int) -> None:
|
|
pending = context.lccov_client._pending_notifications
|
|
assert len(pending) == count, (
|
|
f"Expected {count} pending notifications, got {len(pending)}"
|
|
)
|
|
|
|
|
|
@then("lccov the client should be initialized")
|
|
def step_client_is_initialized(context: Any) -> None:
|
|
assert context.lccov_client.is_initialized
|
|
|
|
|
|
@then('lccov server_capabilities should include "{key}"')
|
|
def step_capabilities_include(context: Any, key: str) -> None:
|
|
caps = context.lccov_client.server_capabilities
|
|
assert key in caps, f"Expected '{key}' in capabilities, got {list(caps.keys())}"
|
|
|
|
|
|
@then('lccov the workspace URI sent should start with "file://"')
|
|
def step_workspace_uri_prefix(context: Any) -> None:
|
|
# Find the initialize request in sent messages
|
|
init_msg = None
|
|
for msg in context.lccov_sent_messages:
|
|
if msg.get("method") == "initialize":
|
|
init_msg = msg
|
|
break
|
|
assert init_msg is not None, "No initialize message found"
|
|
root_uri = init_msg["params"]["rootUri"]
|
|
assert root_uri.startswith("file://"), (
|
|
f"rootUri does not start with file://: {root_uri}"
|
|
)
|
|
|
|
|
|
@then('lccov the workspace URI sent should be "{expected}"')
|
|
def step_workspace_uri_exact(context: Any, expected: str) -> None:
|
|
init_msg = None
|
|
for msg in context.lccov_sent_messages:
|
|
if msg.get("method") == "initialize":
|
|
init_msg = msg
|
|
break
|
|
assert init_msg is not None, "No initialize message found"
|
|
root_uri = init_msg["params"]["rootUri"]
|
|
assert root_uri == expected, f"Expected rootUri={expected}, got {root_uri}"
|
|
|
|
|
|
@then("lccov the sent params should include initializationOptions")
|
|
def step_params_have_init_options(context: Any) -> None:
|
|
init_msg = None
|
|
for msg in context.lccov_sent_messages:
|
|
if msg.get("method") == "initialize":
|
|
init_msg = msg
|
|
break
|
|
assert init_msg is not None, "No initialize message found"
|
|
assert "initializationOptions" in init_msg["params"]
|
|
assert init_msg["params"]["initializationOptions"] == {"custom": True}
|
|
|
|
|
|
@then("lccov no request should have been sent to the transport")
|
|
def step_no_messages_sent(context: Any) -> None:
|
|
assert len(context.lccov_sent_messages) == 0, (
|
|
f"Expected no messages, got {len(context.lccov_sent_messages)}"
|
|
)
|
|
|
|
|
|
@then("lccov the client should not be initialized after shutdown")
|
|
def step_client_not_initialized(context: Any) -> None:
|
|
assert not context.lccov_client.is_initialized
|
|
|
|
|
|
@then("lccov the transport should have received shutdown and exit messages")
|
|
def step_shutdown_and_exit_sent(context: Any) -> None:
|
|
methods = [m.get("method") for m in context.lccov_sent_messages]
|
|
assert "shutdown" in methods, f"No shutdown message found in {methods}"
|
|
assert "exit" in methods, f"No exit message found in {methods}"
|
|
|
|
|
|
@then("lccov the transport should have received a didOpen notification")
|
|
def step_did_open_sent(context: Any) -> None:
|
|
assert len(context.lccov_sent_messages) >= 1
|
|
msg = context.lccov_sent_messages[-1]
|
|
assert msg["method"] == "textDocument/didOpen"
|
|
assert "params" in msg
|
|
td = msg["params"]["textDocument"]
|
|
assert td["uri"] == "file:///test.py"
|
|
assert td["languageId"] == "python"
|
|
assert td["version"] == 1
|
|
assert td["text"] == "print('hello')"
|
|
|
|
|
|
@then("lccov the transport should have received a didClose notification")
|
|
def step_did_close_sent(context: Any) -> None:
|
|
found = any(
|
|
m.get("method") == "textDocument/didClose" for m in context.lccov_sent_messages
|
|
)
|
|
assert found, "No textDocument/didClose message found"
|
|
|
|
|
|
@then('lccov the diagnostics cache for "{uri}" should be empty')
|
|
def step_diagnostics_empty(context: Any, uri: str) -> None:
|
|
assert uri not in context.lccov_client._diagnostics
|
|
|
|
|
|
@then("lccov the returned diagnostics list should have {count:d} item")
|
|
def step_returned_diag_count_singular(context: Any, count: int) -> None:
|
|
assert len(context.lccov_diag_result) == count, (
|
|
f"Expected {count} diagnostics, got {len(context.lccov_diag_result)}"
|
|
)
|
|
|
|
|
|
@then("lccov the returned diagnostics list should have {count:d} items")
|
|
def step_returned_diag_count(context: Any, count: int) -> None:
|
|
assert len(context.lccov_diag_result) == count, (
|
|
f"Expected {count} diagnostics, got {len(context.lccov_diag_result)}"
|
|
)
|
|
|
|
|
|
@then("lccov the completions should have {count:d} items")
|
|
def step_completions_count(context: Any, count: int) -> None:
|
|
assert len(context.lccov_completions) == count, (
|
|
f"Expected {count} completions, got {len(context.lccov_completions)}"
|
|
)
|