from __future__ import annotations import json from typing import Any from behave import given, then, when from behave.runner import Context from pydantic import ValidationError as PydanticValidationError from cleveragents.lsp.models import ( LspCapability, LspServerConfig, LspTransport, ) from cleveragents.lsp.registry import LspRegistry # --------------------------------------------------------------------------- # LspTransport enum steps # --------------------------------------------------------------------------- @given("the LspTransport enum is defined") def step_transport_enum_defined(context: Context) -> None: context.lsp_transport_enum = LspTransport @then('the LspTransport enum should have "{value}" value') def step_transport_enum_value(context: Context, value: str) -> None: assert value in [t.value for t in LspTransport], ( f"LspTransport does not have value '{value}'" ) # --------------------------------------------------------------------------- # Config creation helpers # --------------------------------------------------------------------------- def _minimal_config(**overrides: Any) -> LspServerConfig: """Create a minimal valid LspServerConfig with overrides.""" defaults: dict[str, Any] = { "name": "local/test-server", "command": "echo", "languages": ["python"], } defaults.update(overrides) return LspServerConfig(**defaults) @given("I create a minimal LspServerConfig") def step_create_minimal_config(context: Context) -> None: context.lsp_cfg = _minimal_config() @given('I create a config with description "{desc}"') def step_create_config_with_desc(context: Context, desc: str) -> None: context.lsp_cfg = _minimal_config(description=desc) @given("I create a config with a description of exactly 1000 characters") def step_create_config_1000_desc(context: Context) -> None: context.lsp_cfg = _minimal_config(description="x" * 1000) @given('I create a config with transport "{transport}"') def step_create_config_with_transport(context: Context, transport: str) -> None: context.lsp_cfg = _minimal_config(transport=transport) @given("I create a config with initialization {init_json}") def step_create_config_with_init(context: Context, init_json: str) -> None: init = json.loads(init_json) context.lsp_cfg = _minimal_config(initialization=init) @given("I create a config with workspace_settings {ws_json}") def step_create_config_with_ws(context: Context, ws_json: str) -> None: ws = json.loads(ws_json) context.lsp_cfg = _minimal_config(workspace_settings=ws) @given("I create a config with all new fields populated") def step_create_config_all_fields(context: Context) -> None: context.lsp_cfg = LspServerConfig( name="local/pyright", description="Pyright for Python type checking", command="pyright-langserver", args=["--stdio"], transport=LspTransport.STDIO, languages=["python"], capabilities=[LspCapability.DIAGNOSTICS, LspCapability.COMPLETIONS], env={"PYRIGHT_CACHE": "/tmp/cache"}, initialization={"python": {"pythonPath": "/usr/bin/python3"}}, workspace_settings={"editor.formatOnSave": True}, ) @given("I create a LspServerConfig with only name, command, and languages") def step_create_config_minimal_only(context: Context) -> None: context.lsp_cfg = LspServerConfig( name="local/minimal", command="echo", languages=["python"], ) # --------------------------------------------------------------------------- # Validation error steps # --------------------------------------------------------------------------- @when("I try to create a config with a description of 1001 characters") def step_create_config_long_desc(context: Context) -> None: try: _minimal_config(description="x" * 1001) context.cfg_error = None except (ValueError, PydanticValidationError) as exc: context.cfg_error = exc @when('I try to create a config with transport "{transport}"') def step_create_config_invalid_transport(context: Context, transport: str) -> None: try: _minimal_config(transport=transport) context.cfg_error = None except (ValueError, PydanticValidationError) as exc: context.cfg_error = exc @when("I try to create a config with initialization as a string") def step_create_config_init_string(context: Context) -> None: try: _minimal_config(initialization="not a dict") context.cfg_error = None except (ValueError, PydanticValidationError) as exc: context.cfg_error = exc @when("I try to create a config with workspace_settings as a string") def step_create_config_ws_string(context: Context) -> None: try: _minimal_config(workspace_settings="not a dict") context.cfg_error = None except (ValueError, PydanticValidationError) as exc: context.cfg_error = exc @then("a ValidationError should be raised for config") def step_config_validation_error(context: Context) -> None: assert context.cfg_error is not None, "Expected a validation error" assert isinstance(context.cfg_error, (ValueError, PydanticValidationError)), ( f"Expected ValidationError, got {type(context.cfg_error).__name__}" ) # --------------------------------------------------------------------------- # Assertion steps # --------------------------------------------------------------------------- @then('the config transport should be "{expected}"') def step_check_transport(context: Context, expected: str) -> None: assert context.lsp_cfg.transport.value == expected @then('the config description should be "{expected}"') def step_check_description(context: Context, expected: str) -> None: assert context.lsp_cfg.description == expected @then("the config description should be empty") def step_check_description_empty(context: Context) -> None: assert context.lsp_cfg.description == "" @then("the config description should have length {length:d}") def step_check_description_length(context: Context, length: int) -> None: assert len(context.lsp_cfg.description) == length @then('the config initialization should have key "{key}"') def step_check_init_key(context: Context, key: str) -> None: assert key in context.lsp_cfg.initialization @then('the config initialization "{key}" value should equal {expected_json}') def step_check_init_value(context: Context, key: str, expected_json: str) -> None: expected = json.loads(expected_json) actual = context.lsp_cfg.initialization[key] assert actual == expected, f"Expected {expected}, got {actual}" @then("the config initialization should be empty") def step_check_init_empty(context: Context) -> None: assert context.lsp_cfg.initialization == {} @then('the config workspace_settings should have key "{key}"') def step_check_ws_key(context: Context, key: str) -> None: assert key in context.lsp_cfg.workspace_settings @then("the config workspace_settings should be empty") def step_check_ws_empty(context: Context) -> None: assert context.lsp_cfg.workspace_settings == {} # --------------------------------------------------------------------------- # Serialization round-trip steps # --------------------------------------------------------------------------- @when("I serialize the config to dict and back") def step_serialize_roundtrip(context: Context) -> None: data = context.lsp_cfg.model_dump(mode="json") context.deserialized_cfg = LspServerConfig.model_validate(data) @then("the deserialized config should match the original exactly") def step_check_roundtrip_exact(context: Context) -> None: assert context.lsp_cfg == context.deserialized_cfg @when("I call to_dict on the config") def step_call_to_dict(context: Context) -> None: context.cfg_dict = context.lsp_cfg.to_dict() @then('the config dict should include "{key}"') def step_cfg_dict_has_key(context: Context, key: str) -> None: assert key in context.cfg_dict, ( f"Key '{key}' not in dict: {list(context.cfg_dict.keys())}" ) # --------------------------------------------------------------------------- # Backward compatibility steps # --------------------------------------------------------------------------- @then("the config should have default description") def step_default_desc(context: Context) -> None: assert context.lsp_cfg.description == "" @then("the config should have default transport") def step_default_transport(context: Context) -> None: assert context.lsp_cfg.transport == LspTransport.STDIO @then("the config should have default initialization") def step_default_init(context: Context) -> None: assert context.lsp_cfg.initialization == {} @then("the config should have default workspace_settings") def step_default_ws(context: Context) -> None: assert context.lsp_cfg.workspace_settings == {} # --------------------------------------------------------------------------- # Registry steps # --------------------------------------------------------------------------- @given("I have a clean LspRegistry") def step_clean_registry(context: Context) -> None: context.lsp_reg = LspRegistry() @when("I register the config in the registry") def step_register_config(context: Context) -> None: context.lsp_reg.register(context.lsp_cfg) @then("the registry should contain the config") def step_registry_contains(context: Context) -> None: assert context.lsp_cfg.name in context.lsp_reg @then("retrieving the config should preserve all new fields") def step_registry_preserves_fields(context: Context) -> None: retrieved = context.lsp_reg.get(context.lsp_cfg.name) assert retrieved is not None assert retrieved == context.lsp_cfg