"""Step definitions for LSP CLI commands coverage tests. Covers all four CLI commands in cleveragents.cli.commands.lsp: - add (register from YAML config) - remove (unregister a server) - list (list registered servers) - show (show server details) Also covers the module-level helpers _reset_registry and _get_registry. """ from __future__ import annotations import json import os import tempfile from behave import given, then, when from behave.runner import Context from typer.testing import CliRunner from cleveragents.cli.commands.lsp import _get_registry, _reset_registry, app from cleveragents.lsp.registry import LspRegistry _runner = CliRunner() # --------------------------------------------------------------------------- # YAML fixtures # --------------------------------------------------------------------------- _VALID_YAML = """\ name: local/test-lsp command: test-langserver args: - "--stdio" languages: - python capabilities: - diagnostics - completions - hover """ _VALID_YAML_2 = """\ name: devops/clangd command: clangd args: - "--background-index" languages: - c - cpp capabilities: - diagnostics - document_symbols """ _VALID_YAML_ENV = """\ name: local/test-lsp-env command: test-langserver-env args: - "--stdio" languages: - python capabilities: - diagnostics env: VIRTUAL_ENV: /tmp/venv PATH: /tmp/venv/bin """ _LIST_YAML = """\ - name: invalid command: invalid """ _BAD_SCHEMA_YAML = """\ name: missing-slash-in-name command: some-command languages: - python capabilities: - diagnostics """ # --------------------------------------------------------------------------- # Helper to write temp YAML # --------------------------------------------------------------------------- def _write_temp_yaml(content: str) -> str: """Write content to a temporary YAML file and return its path.""" fd, path = tempfile.mkstemp(suffix=".yaml") with os.fdopen(fd, "w") as fh: fh.write(content) return path # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("a clean LSP CLI test environment") def step_clean_lsp_cli_env(context: Context) -> None: _reset_registry() context.lsp_cli_result = None context.lsp_yaml_path = None context.lsp_invalid_yaml_path = None context.lsp_bad_schema_yaml_path = None context.lsp_env_yaml_path = None context._cleanup_paths: list[str] = [] # --------------------------------------------------------------------------- # Given steps # --------------------------------------------------------------------------- @given("a valid LSP server YAML config file") def step_valid_lsp_yaml(context: Context) -> None: path = _write_temp_yaml(_VALID_YAML) context.lsp_yaml_path = path context._cleanup_paths.append(path) @given("a YAML config file containing a list instead of a mapping") def step_invalid_yaml_list(context: Context) -> None: path = _write_temp_yaml(_LIST_YAML) context.lsp_invalid_yaml_path = path context._cleanup_paths.append(path) @given("a YAML config file with invalid schema fields") def step_bad_schema_yaml(context: Context) -> None: path = _write_temp_yaml(_BAD_SCHEMA_YAML) context.lsp_bad_schema_yaml_path = path context._cleanup_paths.append(path) @given("a valid LSP server YAML config file with env vars") def step_valid_lsp_yaml_env(context: Context) -> None: path = _write_temp_yaml(_VALID_YAML_ENV) context.lsp_env_yaml_path = path context._cleanup_paths.append(path) @given("the LSP server has been registered via CLI") def step_lsp_server_registered(context: Context) -> None: result = _runner.invoke(app, ["add", "--config", context.lsp_yaml_path]) assert result.exit_code == 0, ( f"Pre-registration failed (exit {result.exit_code}): {result.output}" ) @given("the LSP server with env has been registered via CLI") def step_lsp_server_env_registered(context: Context) -> None: result = _runner.invoke(app, ["add", "--config", context.lsp_env_yaml_path]) assert result.exit_code == 0, ( f"Pre-registration failed (exit {result.exit_code}): {result.output}" ) @given("multiple LSP servers have been registered via CLI") def step_multiple_lsp_servers_registered(context: Context) -> None: path1 = _write_temp_yaml(_VALID_YAML) path2 = _write_temp_yaml(_VALID_YAML_2) context._cleanup_paths.extend([path1, path2]) result1 = _runner.invoke(app, ["add", "--config", path1]) assert result1.exit_code == 0, ( f"Registration 1 failed (exit {result1.exit_code}): {result1.output}" ) result2 = _runner.invoke(app, ["add", "--config", path2]) assert result2.exit_code == 0, ( f"Registration 2 failed (exit {result2.exit_code}): {result2.output}" ) # --------------------------------------------------------------------------- # When steps — lsp add # --------------------------------------------------------------------------- @when("I run lsp CLI add with --config pointing to the YAML file") def step_run_lsp_add(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_yaml_path] ) @when("I run lsp CLI add with --config pointing to a non-existent file") def step_run_lsp_add_missing(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", "/tmp/nonexistent_lsp_config_abcdef.yaml"] ) @when("I run lsp CLI add with --config pointing to the invalid YAML file") def step_run_lsp_add_invalid_yaml(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_invalid_yaml_path] ) @when("I run lsp CLI add with --config pointing to the bad schema YAML file") def step_run_lsp_add_bad_schema(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_bad_schema_yaml_path] ) @when("I run lsp CLI add with --config and --update") def step_run_lsp_add_update(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_yaml_path, "--update"] ) @when("I run lsp CLI add with --config and -u short form") def step_run_lsp_add_u_short(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_yaml_path, "-u"] ) @when("I run lsp CLI add with --config and --format json") def step_run_lsp_add_json(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_yaml_path, "--format", "json"] ) @when("I run lsp CLI add with --config without --update") def step_run_lsp_add_no_update(context: Context) -> None: context.lsp_cli_result = _runner.invoke( app, ["add", "--config", context.lsp_yaml_path] ) # --------------------------------------------------------------------------- # When steps — lsp remove # --------------------------------------------------------------------------- @when('I run lsp CLI remove with name "{name}" and --yes') def step_run_lsp_remove(context: Context, name: str) -> None: context.lsp_cli_result = _runner.invoke(app, ["remove", name, "--yes"]) @when('I run lsp CLI remove with name "{name}" and --yes and --format json') def step_run_lsp_remove_json(context: Context, name: str) -> None: context.lsp_cli_result = _runner.invoke( app, ["remove", name, "--yes", "--format", "json"] ) # --------------------------------------------------------------------------- # When steps — lsp list # --------------------------------------------------------------------------- @when("I run lsp CLI list") def step_run_lsp_list(context: Context) -> None: context.lsp_cli_result = _runner.invoke(app, ["list"]) @when('I run lsp CLI list with --namespace "{ns}"') def step_run_lsp_list_namespace(context: Context, ns: str) -> None: context.lsp_cli_result = _runner.invoke(app, ["list", "--namespace", ns]) @when("I run lsp CLI list with --format json") def step_run_lsp_list_json(context: Context) -> None: context.lsp_cli_result = _runner.invoke(app, ["list", "--format", "json"]) @when('I run lsp CLI list with --language "{lang}"') def step_run_lsp_list_language(context: Context, lang: str) -> None: context.lsp_cli_result = _runner.invoke(app, ["list", "--language", lang]) # --------------------------------------------------------------------------- # When steps — lsp show # --------------------------------------------------------------------------- @when('I run lsp CLI show with name "{name}"') def step_run_lsp_show(context: Context, name: str) -> None: context.lsp_cli_result = _runner.invoke(app, ["show", name]) @when('I run lsp CLI show with name "{name}" and --format json') def step_run_lsp_show_json(context: Context, name: str) -> None: context.lsp_cli_result = _runner.invoke(app, ["show", name, "--format", "json"]) # --------------------------------------------------------------------------- # When steps — helpers # --------------------------------------------------------------------------- @when("I call _reset_registry and then _get_registry") def step_reset_and_get_registry(context: Context) -> None: _reset_registry() context.lsp_new_registry = _get_registry() # --------------------------------------------------------------------------- # Then assertions # --------------------------------------------------------------------------- @then("the lsp CLI command should succeed") def step_lsp_cli_success(context: Context) -> None: result = context.lsp_cli_result assert result is not None, "No CLI result captured" assert result.exit_code == 0, ( f"Expected exit code 0, got {result.exit_code}. Output:\n{result.output}" ) @then("the lsp CLI command should abort") def step_lsp_cli_abort(context: Context) -> None: result = context.lsp_cli_result assert result is not None, "No CLI result captured" assert result.exit_code != 0, ( f"Expected non-zero exit code, got {result.exit_code}. Output:\n{result.output}" ) @then('the lsp CLI output should contain "{text}"') def step_lsp_cli_output_contains(context: Context, text: str) -> None: result = context.lsp_cli_result assert result is not None, "No CLI result captured" assert text in result.output, f"Expected '{text}' in output. Got:\n{result.output}" @then('the lsp CLI output should not contain "{text}"') def step_lsp_cli_output_not_contains(context: Context, text: str) -> None: result = context.lsp_cli_result assert result is not None, "No CLI result captured" assert text not in result.output, ( f"Did not expect '{text}' in output. Got:\n{result.output}" ) @then("the lsp CLI output should be valid JSON") def step_lsp_cli_output_valid_json(context: Context) -> None: result = context.lsp_cli_result assert result is not None, "No CLI result captured" try: parsed = json.loads(result.output) assert parsed is not None except json.JSONDecodeError as exc: raise AssertionError( f"Output is not valid JSON: {exc}\nOutput:\n{result.output}" ) from exc @then("a new LspRegistry instance should be returned") def step_new_registry_instance(context: Context) -> None: assert context.lsp_new_registry is not None assert isinstance(context.lsp_new_registry, LspRegistry) @then("the new registry should be empty") def step_new_registry_empty(context: Context) -> None: assert len(context.lsp_new_registry) == 0, ( f"Expected empty registry, got {len(context.lsp_new_registry)} servers" )