diff --git a/.gitignore b/.gitignore index e97bc1f..5606482 100644 --- a/.gitignore +++ b/.gitignore @@ -1,5 +1,8 @@ # For this project envs/ +.augment/ +.github/instructions/ +Laghari/ # vvvvvvv Default Python Ignore vvvvvvvv # Byte-compiled / optimized / DLL files diff --git a/docs/articles/users/CLIReference.md b/docs/articles/users/CLIReference.md index d01f2b8..9b3e810 100644 --- a/docs/articles/users/CLIReference.md +++ b/docs/articles/users/CLIReference.md @@ -346,7 +346,7 @@ Configure an MCP server on a specific host platform. Syntax: -`hatch mcp configure --host (--command CMD | --url URL) [--args ARGS] [--env ENV] [--headers HEADERS] [--dry-run] [--auto-approve] [--no-backup]` +`hatch mcp configure --host (--command CMD | --url URL) [--args ARGS] [--env ENV] [--header HEADER] [--dry-run] [--auto-approve] [--no-backup]` | Argument / Flag | Type | Description | Default | |---:|---|---|---| @@ -356,7 +356,7 @@ Syntax: | `--url` | string | URL for remote MCP servers (mutually exclusive with --command) | none | | `--args` | multiple | Arguments for MCP server command (only with --command) | none | | `--env` | string | Environment variables format: KEY=VALUE (can be used multiple times) | none | -| `--headers` | string | HTTP headers format: KEY=VALUE (only with --url) | none | +| `--header` | string | HTTP headers format: KEY=VALUE (only with --url) | none | | `--dry-run` | flag | Preview configuration without applying changes | false | | `--auto-approve` | flag | Skip confirmation prompts | false | | `--no-backup` | flag | Skip backup creation before configuration | false | @@ -389,7 +389,7 @@ Configure MCP server 'my-server' on host 'claude-desktop'? [y/N]: y **Example - Remote Server Configuration**: ```bash -$ hatch mcp configure api-server --host claude-desktop --url https://api.example.com --headers Auth=token +$ hatch mcp configure api-server --host claude-desktop --url https://api.example.com --header Auth=token Server 'api-server' created for host 'claude-desktop': name: UPDATED None --> 'api-server' diff --git a/docs/articles/users/MCPHostConfiguration.md b/docs/articles/users/MCPHostConfiguration.md index 66aa057..edc3037 100644 --- a/docs/articles/users/MCPHostConfiguration.md +++ b/docs/articles/users/MCPHostConfiguration.md @@ -87,7 +87,7 @@ hatch mcp remove host claude-desktop **Important**: Each server must be configured as either local (using `--command`) or remote (using `--url`), but not both. These options are mutually exclusive: - **Local servers**: Use `--command` and optionally `--args` and `--env` -- **Remote servers**: Use `--url` and optionally `--headers` +- **Remote servers**: Use `--url` and optionally `--header` Attempting to use both `--command` and `--url` will result in an error. diff --git a/hatch/cli_hatch.py b/hatch/cli_hatch.py index b51fd6f..73288c6 100644 --- a/hatch/cli_hatch.py +++ b/hatch/cli_hatch.py @@ -10,6 +10,7 @@ import argparse import json import logging +import shlex import sys from pathlib import Path from typing import Optional, List @@ -567,13 +568,13 @@ def parse_env_vars(env_list: Optional[list]) -> dict: return env_dict -def parse_headers(headers_list: Optional[list]) -> dict: +def parse_header(header_list: Optional[list]) -> dict: """Parse HTTP headers from command line format.""" - if not headers_list: + if not header_list: return {} headers_dict = {} - for header in headers_list: + for header in header_list: if '=' not in header: print(f"Warning: Invalid header format '{header}'. Expected KEY=VALUE") continue @@ -582,7 +583,7 @@ def parse_headers(headers_list: Optional[list]) -> dict: return headers_dict -def parse_inputs(inputs_list: Optional[list]) -> Optional[list]: +def parse_input(input_list: Optional[list]) -> Optional[list]: """Parse VS Code input variable definitions from command line format. Format: type,id,description[,password=true] @@ -591,11 +592,11 @@ def parse_inputs(inputs_list: Optional[list]) -> Optional[list]: Returns: List of input variable definition dictionaries, or None if no inputs provided. """ - if not inputs_list: + if not input_list: return None parsed_inputs = [] - for input_str in inputs_list: + for input_str in input_list: parts = [p.strip() for p in input_str.split(',')] if len(parts) < 3: print(f"Warning: Invalid input format '{input_str}'. Expected: type,id,description[,password=true]") @@ -617,11 +618,11 @@ def parse_inputs(inputs_list: Optional[list]) -> Optional[list]: def handle_mcp_configure(host: str, server_name: str, command: str, args: list, env: Optional[list] = None, url: Optional[str] = None, - headers: Optional[list] = None, timeout: Optional[int] = None, + header: Optional[list] = None, timeout: Optional[int] = None, trust: bool = False, cwd: Optional[str] = None, env_file: Optional[str] = None, http_url: Optional[str] = None, include_tools: Optional[list] = None, exclude_tools: Optional[list] = None, - inputs: Optional[list] = None, no_backup: bool = False, + input: Optional[list] = None, no_backup: bool = False, dry_run: bool = False, auto_approve: bool = False): """Handle 'hatch mcp configure' command with ALL host-specific arguments. @@ -637,23 +638,41 @@ def handle_mcp_configure(host: str, server_name: str, command: str, args: list, print(f"Error: Invalid host '{host}'. Supported hosts: {[h.value for h in MCPHostType]}") return 1 + # Validate Claude Desktop/Code transport restrictions (Issue 2) + if host_type in (MCPHostType.CLAUDE_DESKTOP, MCPHostType.CLAUDE_CODE): + if url is not None: + print(f"Error: {host} does not support remote servers (--url). Only local servers with --command are supported.") + return 1 + # Validate argument dependencies - if command and headers: - print("Error: --headers can only be used with --url (remote servers), not with --command (local servers)") + if command and header: + print("Error: --header can only be used with --url or --http-url (remote servers), not with --command (local servers)") return 1 - if url and args: - print("Error: --args can only be used with --command (local servers), not with --url (remote servers)") + if (url or http_url) and args: + print("Error: --args can only be used with --command (local servers), not with --url or --http-url (remote servers)") return 1 # NOTE: We do NOT validate host-specific arguments here. # The reporting system will show unsupported fields as "UNSUPPORTED" in the conversion report. # This allows users to see which fields are not supported by their target host without blocking the operation. + # Check if server exists (for partial update support) + manager = MCPHostConfigurationManager() + existing_config = manager.get_server_config(host, server_name) + is_update = existing_config is not None + + # Conditional validation: Create requires command OR url OR http_url, update does not + if not is_update: + # Create operation: require command, url, or http_url + if not command and not url and not http_url: + print(f"Error: When creating a new server, you must provide either --command (for local servers), --url (for SSE remote servers), or --http-url (for HTTP remote servers, Gemini only)") + return 1 + # Parse environment variables, headers, and inputs env_dict = parse_env_vars(env) - headers_dict = parse_headers(headers) - inputs_list = parse_inputs(inputs) + headers_dict = parse_header(header) + inputs_list = parse_input(input) # Create Omni configuration (universal model) # Only include fields that have actual values to ensure model_dump(exclude_unset=True) works correctly @@ -662,12 +681,24 @@ def handle_mcp_configure(host: str, server_name: str, command: str, args: list, if command is not None: omni_config_data['command'] = command if args is not None: - omni_config_data['args'] = args + # Process args with shlex.split() to handle quoted strings (Issue 4) + processed_args = [] + for arg in args: + if arg: # Skip empty strings + try: + # Split quoted strings into individual arguments + split_args = shlex.split(arg) + processed_args.extend(split_args) + except ValueError as e: + # Handle invalid quotes gracefully + print(f"Warning: Invalid quote in argument '{arg}': {e}") + processed_args.append(arg) + omni_config_data['args'] = processed_args if processed_args else None if env_dict: omni_config_data['env'] = env_dict if url is not None: omni_config_data['url'] = url - if url and headers_dict: + if headers_dict: omni_config_data['headers'] = headers_dict # Host-specific fields (Gemini) @@ -692,6 +723,29 @@ def handle_mcp_configure(host: str, server_name: str, command: str, args: list, if inputs_list is not None: omni_config_data['inputs'] = inputs_list + # Partial update merge logic + if is_update: + # Merge with existing configuration + existing_data = existing_config.model_dump(exclude_unset=True, exclude={'name'}) + + # Handle command/URL/httpUrl switching behavior + # If switching from command to URL or httpUrl: clear command-based fields + if (url is not None or http_url is not None) and existing_config.command is not None: + existing_data.pop('command', None) + existing_data.pop('args', None) + existing_data.pop('type', None) # Clear type field when switching transports (Issue 1) + + # If switching from URL/httpUrl to command: clear URL-based fields + if command is not None and (existing_config.url is not None or getattr(existing_config, 'httpUrl', None) is not None): + existing_data.pop('url', None) + existing_data.pop('httpUrl', None) + existing_data.pop('headers', None) + existing_data.pop('type', None) # Clear type field when switching transports (Issue 1) + + # Merge: new values override existing values + merged_data = {**existing_data, **omni_config_data} + omni_config_data = merged_data + # Create Omni model omni_config = MCPServerConfigOmni(**omni_config_data) @@ -706,10 +760,11 @@ def handle_mcp_configure(host: str, server_name: str, command: str, args: list, # Generate conversion report report = generate_conversion_report( - operation='create', + operation='update' if is_update else 'create', server_name=server_name, target_host=host_type, omni=omni_config, + old_config=existing_config if is_update else None, dry_run=dry_run ) @@ -1208,17 +1263,17 @@ def main(): # Create mutually exclusive group for server type server_type_group = mcp_configure_parser.add_mutually_exclusive_group(required=True) server_type_group.add_argument("--command", dest="server_command", help="Command to execute the MCP server (for local servers)") - server_type_group.add_argument("--url", help="Server URL for remote MCP servers") + server_type_group.add_argument("--url", help="Server URL for remote MCP servers (SSE transport)") + server_type_group.add_argument("--http-url", help="HTTP streaming endpoint URL (Gemini only)") mcp_configure_parser.add_argument("--args", nargs="*", help="Arguments for the MCP server command (only with --command)") mcp_configure_parser.add_argument("--env-var", action="append", help="Environment variables (format: KEY=VALUE)") - mcp_configure_parser.add_argument("--headers", action="append", help="HTTP headers for remote servers (format: KEY=VALUE, only with --url)") + mcp_configure_parser.add_argument("--header", action="append", help="HTTP headers for remote servers (format: KEY=VALUE, only with --url)") # Host-specific arguments (Gemini) mcp_configure_parser.add_argument("--timeout", type=int, help="Request timeout in milliseconds (Gemini)") mcp_configure_parser.add_argument("--trust", action="store_true", help="Bypass tool call confirmations (Gemini)") mcp_configure_parser.add_argument("--cwd", help="Working directory for stdio transport (Gemini)") - mcp_configure_parser.add_argument("--http-url", help="HTTP streaming endpoint URL (Gemini)") mcp_configure_parser.add_argument("--include-tools", nargs="*", help="Tool allowlist - only these tools will be available (Gemini)") mcp_configure_parser.add_argument("--exclude-tools", nargs="*", help="Tool blocklist - these tools will be excluded (Gemini)") @@ -1226,7 +1281,7 @@ def main(): mcp_configure_parser.add_argument("--env-file", help="Path to environment file (Cursor, VS Code, LM Studio)") # Host-specific arguments (VS Code) - mcp_configure_parser.add_argument("--inputs", action="append", help="Input variable definitions in format: type,id,description[,password=true] (VS Code)") + mcp_configure_parser.add_argument("--input", action="append", help="Input variable definitions in format: type,id,description[,password=true] (VS Code)") mcp_configure_parser.add_argument("--no-backup", action="store_true", help="Skip backup creation before configuration") mcp_configure_parser.add_argument("--dry-run", action="store_true", help="Preview configuration without execution") @@ -2022,11 +2077,11 @@ def main(): elif args.mcp_command == "configure": return handle_mcp_configure( args.host, args.server_name, args.server_command, args.args, - getattr(args, 'env_var', None), args.url, args.headers, + getattr(args, 'env_var', None), args.url, args.header, getattr(args, 'timeout', None), getattr(args, 'trust', False), getattr(args, 'cwd', None), getattr(args, 'env_file', None), getattr(args, 'http_url', None), getattr(args, 'include_tools', None), - getattr(args, 'exclude_tools', None), getattr(args, 'inputs', None), + getattr(args, 'exclude_tools', None), getattr(args, 'input', None), args.no_backup, args.dry_run, args.auto_approve ) diff --git a/hatch/mcp_host_config/host_management.py b/hatch/mcp_host_config/host_management.py index 1592f7f..56c8b5b 100644 --- a/hatch/mcp_host_config/host_management.py +++ b/hatch/mcp_host_config/host_management.py @@ -180,8 +180,32 @@ def configure_server(self, server_config: MCPServerConfig, hostname=hostname, error_message=str(e) ) - - def remove_server(self, server_name: str, hostname: str, + + def get_server_config(self, hostname: str, server_name: str) -> Optional[MCPServerConfig]: + """ + Get existing server configuration from host. + + Args: + hostname: The MCP host to query (e.g., 'claude-desktop', 'cursor') + server_name: Name of the server to retrieve + + Returns: + MCPServerConfig if server exists, None otherwise + """ + try: + host_type = MCPHostType(hostname) + strategy = self.host_registry.get_strategy(host_type) + current_config = strategy.read_configuration() + + if server_name in current_config.servers: + return current_config.servers[server_name] + return None + + except Exception as e: + logger.debug(f"Failed to retrieve server config for {server_name} on {hostname}: {e}") + return None + + def remove_server(self, server_name: str, hostname: str, no_backup: bool = False) -> ConfigurationResult: """Remove MCP server from specified host.""" try: diff --git a/hatch/mcp_host_config/models.py b/hatch/mcp_host_config/models.py index a713ed8..b265370 100644 --- a/hatch/mcp_host_config/models.py +++ b/hatch/mcp_host_config/models.py @@ -360,7 +360,14 @@ class MCPServerConfigBase(BaseModel): @model_validator(mode='after') def validate_transport(self) -> 'MCPServerConfigBase': - """Validate transport configuration using type field.""" + """Validate transport configuration using type field. + + Note: Gemini subclass overrides this with dual-transport support. + """ + # Skip validation for Gemini which has its own dual-transport validator + if self.__class__.__name__ == 'MCPServerConfigGemini': + return self + # Check mutual exclusion - command and url cannot both be set if self.command is not None and self.url is not None: raise ValueError( @@ -413,6 +420,45 @@ class MCPServerConfigGemini(MCPServerConfigBase): oauth_audiences: Optional[List[str]] = Field(None, description="OAuth audiences") authProviderType: Optional[str] = Field(None, description="Authentication provider type") + @model_validator(mode='after') + def validate_gemini_dual_transport(self): + """Override transport validation to support Gemini's dual-transport capability. + + Gemini supports both: + - SSE transport with 'url' field + - HTTP transport with 'httpUrl' field + + Validates that: + 1. Either url or httpUrl is provided (not both) + 2. Type field matches the transport being used + """ + # Check if both url and httpUrl are provided + if self.url is not None and self.httpUrl is not None: + raise ValueError("Cannot specify both 'url' and 'httpUrl' - choose one transport") + + # Validate based on type + if self.type == "stdio": + if not self.command: + raise ValueError("'command' is required for stdio transport") + elif self.type == "sse": + if not self.url: + raise ValueError("'url' is required for sse transport") + elif self.type == "http": + if not self.httpUrl: + raise ValueError("'httpUrl' is required for http transport") + elif self.type is None: + # Infer type from fields if not specified + if self.command: + self.type = "stdio" + elif self.url: + self.type = "sse" # default to sse for url + elif self.httpUrl: + self.type = "http" # http for httpUrl + else: + raise ValueError("Either 'command', 'url', or 'httpUrl' must be provided") + + return self + @classmethod def from_omni(cls, omni: 'MCPServerConfigOmni') -> 'MCPServerConfigGemini': """Convert Omni model to Gemini-specific model using Pydantic APIs.""" diff --git a/tests/test_mcp_cli_all_host_specific_args.py b/tests/test_mcp_cli_all_host_specific_args.py index 20539da..2026fc0 100644 --- a/tests/test_mcp_cli_all_host_specific_args.py +++ b/tests/test_mcp_cli_all_host_specific_args.py @@ -11,7 +11,7 @@ from unittest.mock import patch, MagicMock from io import StringIO -from hatch.cli_hatch import handle_mcp_configure, parse_inputs +from hatch.cli_hatch import handle_mcp_configure, parse_input from hatch.mcp_host_config import MCPHostType from hatch.mcp_host_config.models import ( MCPServerConfigGemini, MCPServerConfigCursor, MCPServerConfigVSCode, @@ -113,7 +113,7 @@ def test_vscode_inputs_on_gemini_show_unsupported(self, mock_stdout, mock_manage server_name='test-server', command='python', args=['server.py'], - inputs=['promptString,api-key,API Key,password=true'], # VS Code-only field + input=['promptString,api-key,API Key,password=true'], # VS Code-only field auto_approve=True ) @@ -129,11 +129,11 @@ def test_vscode_inputs_on_gemini_show_unsupported(self, mock_stdout, mock_manage class TestVSCodeInputsParsing(unittest.TestCase): """Test VS Code inputs parsing.""" - def test_parse_inputs_basic(self): + def test_parse_input_basic(self): """Test basic input parsing.""" - inputs_list = ['promptString,api-key,GitHub Personal Access Token'] - result = parse_inputs(inputs_list) - + input_list = ['promptString,api-key,GitHub Personal Access Token'] + result = parse_input(input_list) + self.assertIsNotNone(result) self.assertEqual(len(result), 1) self.assertEqual(result[0]['type'], 'promptString') @@ -141,34 +141,34 @@ def test_parse_inputs_basic(self): self.assertEqual(result[0]['description'], 'GitHub Personal Access Token') self.assertNotIn('password', result[0]) - def test_parse_inputs_with_password(self): + def test_parse_input_with_password(self): """Test input parsing with password flag.""" - inputs_list = ['promptString,api-key,API Key,password=true'] - result = parse_inputs(inputs_list) - + input_list = ['promptString,api-key,API Key,password=true'] + result = parse_input(input_list) + self.assertIsNotNone(result) self.assertEqual(len(result), 1) self.assertEqual(result[0]['password'], True) - def test_parse_inputs_multiple(self): + def test_parse_input_multiple(self): """Test parsing multiple inputs.""" - inputs_list = [ + input_list = [ 'promptString,api-key,API Key,password=true', 'promptString,db-url,Database URL' ] - result = parse_inputs(inputs_list) - + result = parse_input(input_list) + self.assertIsNotNone(result) self.assertEqual(len(result), 2) - def test_parse_inputs_none(self): + def test_parse_input_none(self): """Test parsing None inputs.""" - result = parse_inputs(None) + result = parse_input(None) self.assertIsNone(result) - def test_parse_inputs_empty(self): + def test_parse_input_empty(self): """Test parsing empty inputs list.""" - result = parse_inputs([]) + result = parse_input([]) self.assertIsNone(result) @@ -191,7 +191,7 @@ def test_vscode_inputs_passed_to_model(self, mock_manager_class): server_name='test-server', command='python', args=['server.py'], - inputs=['promptString,api-key,API Key,password=true'], + input=['promptString,api-key,API Key,password=true'], auto_approve=True ) diff --git a/tests/test_mcp_cli_direct_management.py b/tests/test_mcp_cli_direct_management.py index cfd4c69..44ddfc6 100644 --- a/tests/test_mcp_cli_direct_management.py +++ b/tests/test_mcp_cli_direct_management.py @@ -19,7 +19,7 @@ from hatch.cli_hatch import ( main, handle_mcp_configure, handle_mcp_remove, handle_mcp_remove_server, - handle_mcp_remove_host, parse_env_vars, parse_headers + handle_mcp_remove_host, parse_env_vars, parse_header ) from hatch.mcp_host_config.models import MCPHostType, MCPServerConfig from wobble import regression_test, integration_test @@ -61,7 +61,7 @@ def test_configure_argument_parsing_with_options(self): test_args = [ 'hatch', 'mcp', 'configure', 'file-server', '--host', 'cursor', '--url', 'http://localhost:8080', '--env-var', 'API_KEY=secret', '--env-var', 'DEBUG=true', - '--headers', 'Authorization=Bearer token', + '--header', 'Authorization=Bearer token', '--no-backup', '--dry-run', '--auto-approve' ] @@ -104,21 +104,21 @@ def test_parse_env_vars(self): mock_print.assert_called() @regression_test - def test_parse_headers(self): + def test_parse_header(self): """Test HTTP headers parsing utility.""" # Valid headers headers_list = ['Authorization=Bearer token', 'Content-Type=application/json'] - result = parse_headers(headers_list) - + result = parse_header(headers_list) + expected = { 'Authorization': 'Bearer token', 'Content-Type': 'application/json' } self.assertEqual(result, expected) - + # Empty list - self.assertEqual(parse_headers(None), {}) - self.assertEqual(parse_headers([]), {}) + self.assertEqual(parse_header(None), {}) + self.assertEqual(parse_header([]), {}) @integration_test(scope="component") def test_configure_invalid_host(self): diff --git a/tests/test_mcp_cli_host_config_integration.py b/tests/test_mcp_cli_host_config_integration.py index f1fd22c..468c074 100644 --- a/tests/test_mcp_cli_host_config_integration.py +++ b/tests/test_mcp_cli_host_config_integration.py @@ -31,7 +31,7 @@ def decorator(func): from hatch.cli_hatch import ( handle_mcp_configure, parse_env_vars, - parse_headers, + parse_header, parse_host_list, ) from hatch.mcp_host_config.models import ( @@ -68,7 +68,7 @@ def test_configure_creates_omni_model_basic(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -90,7 +90,7 @@ def test_configure_creates_omni_with_env_vars(self): args=['server.py'], env=['API_KEY=secret', 'DEBUG=true'], url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -105,13 +105,13 @@ def test_configure_creates_omni_with_headers(self): with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager: with patch('hatch.cli_hatch.request_confirmation', return_value=False): result = handle_mcp_configure( - host='claude-desktop', + host='gemini', # Use gemini which supports remote servers server_name='test-server', command=None, args=None, env=None, url='https://api.example.com', - headers=['Authorization=Bearer token', 'Content-Type=application/json'], + header=['Authorization=Bearer token', 'Content-Type=application/json'], no_backup=True, dry_run=False, auto_approve=False @@ -126,13 +126,13 @@ def test_configure_creates_omni_remote_server(self): with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager: with patch('hatch.cli_hatch.request_confirmation', return_value=False): result = handle_mcp_configure( - host='claude-desktop', + host='gemini', # Use gemini which supports remote servers server_name='remote-server', command=None, args=None, env=None, url='https://api.example.com', - headers=['Auth=token'], + header=['Auth=token'], no_backup=True, dry_run=False, auto_approve=False @@ -154,7 +154,7 @@ def test_configure_omni_with_all_universal_fields(self): args=['server.py', '--port', '8080'], env=['API_KEY=secret', 'DEBUG=true', 'LOG_LEVEL=info'], url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -176,7 +176,7 @@ def test_configure_omni_with_optional_fields_none(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -202,7 +202,7 @@ def test_configure_uses_host_model_registry(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -224,7 +224,7 @@ def test_configure_calls_from_omni_conversion(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -250,7 +250,7 @@ def test_configure_passes_host_specific_model_to_manager(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -282,7 +282,7 @@ def test_configure_dry_run_displays_report_only(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=True, auto_approve=False @@ -291,8 +291,9 @@ def test_configure_dry_run_displays_report_only(self): # Verify the function executed without errors self.assertEqual(result, 0) - # Verify MCPHostConfigurationManager was not instantiated (no actual configuration) - mock_manager.assert_not_called() + # Verify MCPHostConfigurationManager.create_server was NOT called (dry-run doesn't persist) + # Note: get_server_config is called to check if server exists, but create_server is not called + mock_manager.return_value.create_server.assert_not_called() class TestHostSpecificArguments(unittest.TestCase): @@ -311,7 +312,7 @@ def test_configure_accepts_all_universal_fields(self): args=['server.py', '--port', '8080'], env=['API_KEY=secret', 'DEBUG=true'], url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -333,7 +334,7 @@ def test_configure_multiple_env_vars(self): args=['server.py'], env=['VAR1=value1', 'VAR2=value2', 'VAR3=value3'], url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -358,7 +359,7 @@ def test_configure_different_hosts(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -382,7 +383,7 @@ def test_configure_invalid_host_type_error(self): args=['server.py'], env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -403,7 +404,7 @@ def test_configure_invalid_field_value_error(self): args=None, # Must be None for remote server env=None, url='not-a-url', # Invalid URL format - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -423,7 +424,7 @@ def test_configure_pydantic_validation_error_handling(self): args=['server.py'], env=None, url=None, - headers=['Auth=token'], # Headers not allowed with command + header=['Auth=token'], # Headers not allowed with command no_backup=True, dry_run=False, auto_approve=False @@ -445,7 +446,7 @@ def test_configure_missing_command_url_error(self): args=None, env=None, url=None, - headers=None, + header=None, no_backup=True, dry_run=False, auto_approve=False @@ -475,7 +476,7 @@ def test_existing_configure_command_still_works(self): args=['-m', 'my_package.server'], env=['API_KEY=secret'], url=None, - headers=None, + header=None, no_backup=False, dry_run=False, auto_approve=False @@ -508,21 +509,21 @@ def test_parse_env_vars_empty(self): self.assertEqual(result, {}) @regression_test - def test_parse_headers_basic(self): + def test_parse_header_basic(self): """Test parsing headers from KEY=VALUE format.""" headers_list = ['Authorization=Bearer token', 'Content-Type=application/json'] - result = parse_headers(headers_list) + result = parse_header(headers_list) expected = {'Authorization': 'Bearer token', 'Content-Type': 'application/json'} self.assertEqual(result, expected) @regression_test - def test_parse_headers_empty(self): + def test_parse_header_empty(self): """Test parsing empty headers list.""" - result = parse_headers(None) + result = parse_header(None) self.assertEqual(result, {}) - result = parse_headers([]) + result = parse_header([]) self.assertEqual(result, {}) @@ -626,6 +627,178 @@ def test_reporting_functions_available(self): self.assertIsNotNone(report) self.assertEqual(report.operation, 'create') + @regression_test + def test_claude_desktop_rejects_url_configuration(self): + """Test Claude Desktop rejects remote server (--url) configurations (Issue 2).""" + with patch('hatch.cli_hatch.print') as mock_print: + result = handle_mcp_configure( + host='claude-desktop', + server_name='remote-server', + command=None, + args=None, + env=None, + url='http://localhost:8080', # Should be rejected + header=None, + no_backup=True, + dry_run=False, + auto_approve=True + ) + + # Validate: Should return error code 1 + self.assertEqual(result, 1) + + # Validate: Error message displayed + error_calls = [call for call in mock_print.call_args_list + if 'Error' in str(call) or 'error' in str(call)] + self.assertTrue(len(error_calls) > 0, "Expected error message to be printed") + + @regression_test + def test_claude_code_rejects_url_configuration(self): + """Test Claude Code (same family) also rejects remote servers (Issue 2).""" + with patch('hatch.cli_hatch.print') as mock_print: + result = handle_mcp_configure( + host='claude-code', + server_name='remote-server', + command=None, + args=None, + env=None, + url='http://localhost:8080', + header=None, + no_backup=True, + dry_run=False, + auto_approve=True + ) + + # Validate: Should return error code 1 + self.assertEqual(result, 1) + + # Validate: Error message displayed + error_calls = [call for call in mock_print.call_args_list + if 'Error' in str(call) or 'error' in str(call)] + self.assertTrue(len(error_calls) > 0, "Expected error message to be printed") + + @regression_test + def test_args_quoted_string_splitting(self): + """Test that quoted strings in --args are properly split (Issue 4).""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager: + with patch('hatch.cli_hatch.request_confirmation', return_value=False): + # Simulate user providing: --args "-r --name aName" + # This arrives as a single string element in the args list + result = handle_mcp_configure( + host='claude-desktop', + server_name='test-server', + command='python', + args=['-r --name aName'], # Single string with quoted content + env=None, + url=None, + header=None, + no_backup=True, + dry_run=False, + auto_approve=False + ) + + # Verify: Should succeed (return 0) + self.assertEqual(result, 0) + + # Verify: MCPServerConfigOmni was created with split args + call_args = mock_manager.return_value.create_server.call_args + if call_args: + omni_config = call_args[1]['omni'] + # Args should be split into 3 elements: ['-r', '--name', 'aName'] + self.assertEqual(omni_config.args, ['-r', '--name', 'aName']) + + @regression_test + def test_args_multiple_quoted_strings(self): + """Test multiple quoted strings in --args are all split correctly (Issue 4).""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager: + with patch('hatch.cli_hatch.request_confirmation', return_value=False): + # Simulate: --args "-r" "--name aName" + result = handle_mcp_configure( + host='claude-desktop', + server_name='test-server', + command='python', + args=['-r', '--name aName'], # Two separate args + env=None, + url=None, + header=None, + no_backup=True, + dry_run=False, + auto_approve=False + ) + + # Verify: Should succeed + self.assertEqual(result, 0) + + # Verify: All args are properly split + call_args = mock_manager.return_value.create_server.call_args + if call_args: + omni_config = call_args[1]['omni'] + # Should be split into: ['-r', '--name', 'aName'] + self.assertEqual(omni_config.args, ['-r', '--name', 'aName']) + + @regression_test + def test_args_empty_string_handling(self): + """Test that empty strings in --args are filtered out (Issue 4).""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager: + with patch('hatch.cli_hatch.request_confirmation', return_value=False): + # Simulate: --args "" "server.py" + result = handle_mcp_configure( + host='claude-desktop', + server_name='test-server', + command='python', + args=['', 'server.py'], # Empty string should be filtered + env=None, + url=None, + header=None, + no_backup=True, + dry_run=False, + auto_approve=False + ) + + # Verify: Should succeed + self.assertEqual(result, 0) + + # Verify: Empty strings are filtered out + call_args = mock_manager.return_value.create_server.call_args + if call_args: + omni_config = call_args[1]['omni'] + # Should only contain 'server.py' + self.assertEqual(omni_config.args, ['server.py']) + + @regression_test + def test_args_invalid_quote_handling(self): + """Test that invalid quotes in --args are handled gracefully (Issue 4).""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager: + with patch('hatch.cli_hatch.request_confirmation', return_value=False): + with patch('hatch.cli_hatch.print') as mock_print: + # Simulate: --args 'unclosed "quote' + result = handle_mcp_configure( + host='claude-desktop', + server_name='test-server', + command='python', + args=['unclosed "quote'], # Invalid quote + env=None, + url=None, + header=None, + no_backup=True, + dry_run=False, + auto_approve=False + ) + + # Verify: Should succeed (graceful fallback) + self.assertEqual(result, 0) + + # Verify: Warning was printed + warning_calls = [call for call in mock_print.call_args_list + if 'Warning' in str(call)] + self.assertTrue(len(warning_calls) > 0, "Expected warning for invalid quote") + + # Verify: Original arg is used as fallback + call_args = mock_manager.return_value.create_server.call_args + if call_args: + omni_config = call_args[1]['omni'] + self.assertIn('unclosed "quote', omni_config.args) + @regression_test def test_cli_handler_signature_compatible(self): """Test that handle_mcp_configure signature is compatible with integration.""" @@ -638,7 +811,7 @@ def test_cli_handler_signature_compatible(self): # Verify expected parameters exist expected_params = [ 'host', 'server_name', 'command', 'args', - 'env', 'url', 'headers', 'no_backup', 'dry_run', 'auto_approve' + 'env', 'url', 'header', 'no_backup', 'dry_run', 'auto_approve' ] for param in expected_params: diff --git a/tests/test_mcp_cli_partial_updates.py b/tests/test_mcp_cli_partial_updates.py new file mode 100644 index 0000000..d20e9a5 --- /dev/null +++ b/tests/test_mcp_cli_partial_updates.py @@ -0,0 +1,859 @@ +""" +Test suite for MCP CLI partial configuration update functionality. + +This module tests the partial configuration update feature that allows users to modify +specific fields without re-specifying entire server configurations. + +Tests cover: +- Server existence detection (get_server_config method) +- Partial update validation (create vs. update logic) +- Field preservation (merge logic) +- Command/URL switching behavior +- End-to-end integration workflows +- Backward compatibility +""" + +import unittest +from unittest.mock import patch, MagicMock, call +import sys +from pathlib import Path + +# Add the parent directory to the path to import hatch modules +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from hatch.mcp_host_config.host_management import MCPHostConfigurationManager +from hatch.mcp_host_config.models import MCPHostType, MCPServerConfig, MCPServerConfigOmni +from hatch.cli_hatch import handle_mcp_configure +from wobble import regression_test, integration_test + + +class TestServerExistenceDetection(unittest.TestCase): + """Test suite for server existence detection (Category A).""" + + @regression_test + def test_get_server_config_exists(self): + """Test A1: get_server_config returns existing server configuration.""" + # Setup: Create a test server configuration + manager = MCPHostConfigurationManager() + + # Mock the strategy to return a configuration with our test server + mock_strategy = MagicMock() + mock_config = MagicMock() + test_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"], + env={"API_KEY": "test_key"} + ) + mock_config.servers = {"test-server": test_server} + mock_strategy.read_configuration.return_value = mock_config + + with patch.object(manager.host_registry, 'get_strategy', return_value=mock_strategy): + # Execute + result = manager.get_server_config("claude-desktop", "test-server") + + # Validate + self.assertIsNotNone(result) + self.assertEqual(result.name, "test-server") + self.assertEqual(result.command, "python") + + @regression_test + def test_get_server_config_not_exists(self): + """Test A2: get_server_config returns None for non-existent server.""" + # Setup: Empty registry + manager = MCPHostConfigurationManager() + + mock_strategy = MagicMock() + mock_config = MagicMock() + mock_config.servers = {} # No servers + mock_strategy.read_configuration.return_value = mock_config + + with patch.object(manager.host_registry, 'get_strategy', return_value=mock_strategy): + # Execute + result = manager.get_server_config("claude-desktop", "non-existent-server") + + # Validate + self.assertIsNone(result) + + @regression_test + def test_get_server_config_invalid_host(self): + """Test A3: get_server_config handles invalid host gracefully.""" + # Setup + manager = MCPHostConfigurationManager() + + # Execute: Invalid host should be handled gracefully + result = manager.get_server_config("invalid-host", "test-server") + + # Validate: Should return None, not raise exception + self.assertIsNone(result) + + +class TestPartialUpdateValidation(unittest.TestCase): + """Test suite for partial update validation (Category B).""" + + @regression_test + def test_configure_update_single_field_timeout(self): + """Test B1: Update single field (timeout) preserves other fields.""" + # Setup: Existing server with timeout=30 + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"], + env={"API_KEY": "test_key"}, + timeout=30 + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Update only timeout (use Gemini which supports timeout) + result = handle_mcp_configure( + host="gemini", + server_name="test-server", + command=None, + args=None, + env=None, + url=None, + header=None, + timeout=60, # Only timeout provided + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: configure_server was called with merged config + mock_manager.configure_server.assert_called_once() + call_args = mock_manager.configure_server.call_args + host_config = call_args[1]['server_config'] + + # Timeout should be updated (Gemini supports timeout) + self.assertEqual(host_config.timeout, 60) + # Other fields should be preserved + self.assertEqual(host_config.command, "python") + self.assertEqual(host_config.args, ["server.py"]) + + @regression_test + def test_configure_update_env_vars_only(self): + """Test B2: Update environment variables only preserves other fields.""" + # Setup: Existing server with env vars + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"], + env={"API_KEY": "old_key"} + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Update only env vars + result = handle_mcp_configure( + host="claude-desktop", + server_name="test-server", + command=None, + args=None, + env=["NEW_KEY=new_value"], # Only env provided + url=None, + header=None, + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: configure_server was called with merged config + mock_manager.configure_server.assert_called_once() + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + + # Env should be updated + self.assertEqual(omni_config.env, {"NEW_KEY": "new_value"}) + # Other fields should be preserved + self.assertEqual(omni_config.command, "python") + self.assertEqual(omni_config.args, ["server.py"]) + + @regression_test + def test_configure_create_requires_command_or_url(self): + """Test B4: Create operation requires command or url.""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = None # Server doesn't exist + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Create without command or url + result = handle_mcp_configure( + host="claude-desktop", + server_name="new-server", + command=None, # No command + args=None, + env=None, + url=None, # No url + header=None, + timeout=60, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should fail with error + self.assertEqual(result, 1) + + # Validate: Error message mentions command or url + mock_print.assert_called() + error_message = str(mock_print.call_args[0][0]) + self.assertIn("command", error_message.lower()) + self.assertIn("url", error_message.lower()) + + @regression_test + def test_configure_update_allows_no_command_url(self): + """Test B5: Update operation allows omitting command/url.""" + # Setup: Existing server with command + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"] + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Update without command or url + result = handle_mcp_configure( + host="claude-desktop", + server_name="test-server", + command=None, # No command + args=None, + env=None, + url=None, # No url + header=None, + timeout=60, # Only timeout + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: Command should be preserved + mock_manager.configure_server.assert_called_once() + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + self.assertEqual(omni_config.command, "python") + + +class TestFieldPreservation(unittest.TestCase): + """Test suite for field preservation verification (Category C).""" + + @regression_test + def test_configure_update_preserves_unspecified_fields(self): + """Test C1: Unspecified fields remain unchanged during update.""" + # Setup: Existing server with multiple fields + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"], + env={"API_KEY": "test_key"}, + timeout=30 + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Update only timeout (use Gemini which supports timeout) + result = handle_mcp_configure( + host="gemini", + server_name="test-server", + command=None, + args=None, + env=None, + url=None, + header=None, + timeout=60, # Only timeout updated + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate + self.assertEqual(result, 0) + call_args = mock_manager.configure_server.call_args + host_config = call_args[1]['server_config'] + + # Timeout updated (Gemini supports timeout) + self.assertEqual(host_config.timeout, 60) + # All other fields preserved + self.assertEqual(host_config.command, "python") + self.assertEqual(host_config.args, ["server.py"]) + self.assertEqual(host_config.env, {"API_KEY": "test_key"}) + + @regression_test + def test_configure_update_dependent_fields(self): + """Test C3+C4: Update dependent fields without parent field.""" + # Scenario 1: Update args without command + existing_cmd_server = MCPServerConfig( + name="cmd-server", + command="python", + args=["old.py"] + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_cmd_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Update args without command + result = handle_mcp_configure( + host="claude-desktop", + server_name="cmd-server", + command=None, # Command not provided + args=["new.py"], # Args updated + env=None, + url=None, + header=None, + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + + # Args updated, command preserved + self.assertEqual(omni_config.args, ["new.py"]) + self.assertEqual(omni_config.command, "python") + + # Scenario 2: Update headers without url + existing_url_server = MCPServerConfig( + name="url-server", + url="http://localhost:8080", + headers={"Authorization": "Bearer old_token"} + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_url_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Update headers without url + result = handle_mcp_configure( + host="claude-desktop", + server_name="url-server", + command=None, + args=None, + env=None, + url=None, # URL not provided + header=["Authorization=Bearer new_token"], # Headers updated + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + + # Headers updated, url preserved + self.assertEqual(omni_config.headers, {"Authorization": "Bearer new_token"}) + self.assertEqual(omni_config.url, "http://localhost:8080") + + +class TestCommandUrlSwitching(unittest.TestCase): + """Test suite for command/URL switching behavior (Category E) [CRITICAL].""" + + @regression_test + def test_configure_switch_command_to_url(self): + """Test E1: Switch from command-based to URL-based server [CRITICAL].""" + # Setup: Existing command-based server + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"], + env={"API_KEY": "test_key"} + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Switch to URL-based (use gemini which supports URL) + result = handle_mcp_configure( + host="gemini", + server_name="test-server", + command=None, + args=None, + env=None, + url="http://localhost:8080", # Provide URL + header=["Authorization=Bearer token"], # Provide headers + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + + # URL-based fields set + self.assertEqual(omni_config.url, "http://localhost:8080") + self.assertEqual(omni_config.headers, {"Authorization": "Bearer token"}) + # Command-based fields cleared + self.assertIsNone(omni_config.command) + self.assertIsNone(omni_config.args) + # Type field updated to 'sse' (Issue 1) + self.assertEqual(omni_config.type, "sse") + + @regression_test + def test_configure_switch_url_to_command(self): + """Test E2: Switch from URL-based to command-based server [CRITICAL].""" + # Setup: Existing URL-based server + existing_server = MCPServerConfig( + name="test-server", + url="http://localhost:8080", + headers={"Authorization": "Bearer token"} + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Switch to command-based (use gemini which supports both) + result = handle_mcp_configure( + host="gemini", + server_name="test-server", + command="node", # Provide command + args=["server.js"], # Provide args + env=None, + url=None, + header=None, + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + + # Command-based fields set + self.assertEqual(omni_config.command, "node") + self.assertEqual(omni_config.args, ["server.js"]) + # URL-based fields cleared + self.assertIsNone(omni_config.url) + self.assertIsNone(omni_config.headers) + # Type field updated to 'stdio' (Issue 1) + self.assertEqual(omni_config.type, "stdio") + + +class TestPartialUpdateIntegration(unittest.TestCase): + """Test suite for end-to-end partial update workflows (Integration Tests).""" + + @integration_test(scope="component") + def test_partial_update_end_to_end_timeout(self): + """Test I1: End-to-end partial update workflow for timeout field.""" + # Setup: Existing server + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"], + timeout=30 + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + with patch('hatch.cli_hatch.generate_conversion_report') as mock_report: + # Mock report to verify UNCHANGED detection + mock_report.return_value = MagicMock() + + # Execute: Full CLI workflow + result = handle_mcp_configure( + host="claude-desktop", + server_name="test-server", + command=None, + args=None, + env=None, + url=None, + header=None, + timeout=60, # Update timeout only + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: Report was generated with old_config for UNCHANGED detection + mock_report.assert_called_once() + call_kwargs = mock_report.call_args[1] + self.assertEqual(call_kwargs['operation'], 'update') + self.assertIsNotNone(call_kwargs.get('old_config')) + + @integration_test(scope="component") + def test_partial_update_end_to_end_switch_type(self): + """Test I2: End-to-end workflow for command/URL switching.""" + # Setup: Existing command-based server + existing_server = MCPServerConfig( + name="test-server", + command="python", + args=["server.py"] + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + with patch('hatch.cli_hatch.generate_conversion_report') as mock_report: + mock_report.return_value = MagicMock() + + # Execute: Switch to URL-based (use gemini which supports URL) + result = handle_mcp_configure( + host="gemini", + server_name="test-server", + command=None, + args=None, + env=None, + url="http://localhost:8080", + header=["Authorization=Bearer token"], + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: Server type switched + call_args = mock_manager.configure_server.call_args + omni_config = call_args[1]['server_config'] + self.assertEqual(omni_config.url, "http://localhost:8080") + self.assertIsNone(omni_config.command) + + +class TestBackwardCompatibility(unittest.TestCase): + """Test suite for backward compatibility (Regression Tests).""" + + @regression_test + def test_existing_create_operation_unchanged(self): + """Test R1: Existing create operations work identically.""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = None # Server doesn't exist + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Create operation with full configuration (use Gemini for timeout support) + result = handle_mcp_configure( + host="gemini", + server_name="new-server", + command="python", + args=["server.py"], + env=["API_KEY=secret"], + url=None, + header=None, + timeout=30, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: Server created with all fields + mock_manager.configure_server.assert_called_once() + call_args = mock_manager.configure_server.call_args + host_config = call_args[1]['server_config'] + self.assertEqual(host_config.command, "python") + self.assertEqual(host_config.args, ["server.py"]) + self.assertEqual(host_config.timeout, 30) + + @regression_test + def test_error_messages_remain_clear(self): + """Test R2: Error messages are clear and helpful (modified).""" + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = None # Server doesn't exist + + with patch('hatch.cli_hatch.print') as mock_print: + # Execute: Create without command or url + result = handle_mcp_configure( + host="claude-desktop", + server_name="new-server", + command=None, # No command + args=None, + env=None, + url=None, # No url + header=None, + timeout=60, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should fail + self.assertEqual(result, 1) + + # Validate: Error message is clear + mock_print.assert_called() + error_message = str(mock_print.call_args[0][0]) + self.assertIn("command", error_message.lower()) + self.assertIn("url", error_message.lower()) + # Should mention this is for creating a new server + self.assertTrue( + "creat" in error_message.lower() or "new" in error_message.lower(), + f"Error message should clarify this is for creating: {error_message}" + ) + + +class TestTypeFieldUpdating(unittest.TestCase): + """Test suite for type field updates during transport switching (Issue 1).""" + + @regression_test + def test_type_field_updates_command_to_url(self): + """Test type field updates from 'stdio' to 'sse' when switching to URL.""" + # Setup: Create existing command-based server with type='stdio' + existing_server = MCPServerConfig( + name="test-server", + type="stdio", + command="python", + args=["server.py"] + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print'): + # Execute: Switch to URL-based configuration + result = handle_mcp_configure( + host='gemini', + server_name='test-server', + command=None, + args=None, + env=None, + url='http://localhost:8080', + header=None, + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: Type field updated to 'sse' + call_args = mock_manager.configure_server.call_args + server_config = call_args.kwargs['server_config'] + self.assertEqual(server_config.type, "sse") + self.assertIsNone(server_config.command) + self.assertEqual(server_config.url, "http://localhost:8080") + + @regression_test + def test_type_field_updates_url_to_command(self): + """Test type field updates from 'sse' to 'stdio' when switching to command.""" + # Setup: Create existing URL-based server with type='sse' + existing_server = MCPServerConfig( + name="test-server", + type="sse", + url="http://localhost:8080", + headers={"Authorization": "Bearer token"} + ) + + with patch('hatch.cli_hatch.MCPHostConfigurationManager') as mock_manager_class: + mock_manager = MagicMock() + mock_manager_class.return_value = mock_manager + mock_manager.get_server_config.return_value = existing_server + mock_manager.configure_server.return_value = MagicMock(success=True) + + with patch('hatch.cli_hatch.print'): + # Execute: Switch to command-based configuration + result = handle_mcp_configure( + host='gemini', + server_name='test-server', + command='python', + args=['server.py'], + env=None, + url=None, + header=None, + timeout=None, + trust=False, + cwd=None, + env_file=None, + http_url=None, + include_tools=None, + exclude_tools=None, + input=None, + no_backup=False, + dry_run=False, + auto_approve=True + ) + + # Validate: Should succeed + self.assertEqual(result, 0) + + # Validate: Type field updated to 'stdio' + call_args = mock_manager.configure_server.call_args + server_config = call_args.kwargs['server_config'] + self.assertEqual(server_config.type, "stdio") + self.assertEqual(server_config.command, "python") + self.assertIsNone(server_config.url) + + +if __name__ == '__main__': + unittest.main() + diff --git a/tests/test_mcp_pydantic_architecture_v4.py b/tests/test_mcp_pydantic_architecture_v4.py index 30233fc..4a332d9 100644 --- a/tests/test_mcp_pydantic_architecture_v4.py +++ b/tests/test_mcp_pydantic_architecture_v4.py @@ -556,6 +556,48 @@ def test_claude_from_omni_with_universal_fields(self): self.assertEqual(claude.env["API_KEY"], "test") +class TestGeminiDualTransport(unittest.TestCase): + """Test suite for Gemini dual-transport validation (Issue 3).""" + + @regression_test + def test_gemini_sse_transport_with_url(self): + """Test Gemini SSE transport uses url field.""" + config = MCPServerConfigGemini( + name="gemini-server", + type="sse", + url="https://api.example.com/mcp" + ) + + self.assertEqual(config.type, "sse") + self.assertEqual(config.url, "https://api.example.com/mcp") + self.assertIsNone(config.httpUrl) + + @regression_test + def test_gemini_http_transport_with_httpUrl(self): + """Test Gemini HTTP transport uses httpUrl field.""" + config = MCPServerConfigGemini( + name="gemini-server", + type="http", + httpUrl="https://api.example.com/mcp" + ) + + self.assertEqual(config.type, "http") + self.assertEqual(config.httpUrl, "https://api.example.com/mcp") + self.assertIsNone(config.url) + + @regression_test + def test_gemini_mutual_exclusion_url_and_httpUrl(self): + """Test Gemini rejects both url and httpUrl simultaneously.""" + with self.assertRaises(ValidationError) as context: + MCPServerConfigGemini( + name="gemini-server", + url="https://api.example.com/sse", + httpUrl="https://api.example.com/http" + ) + + self.assertIn("Cannot specify both 'url' and 'httpUrl'", str(context.exception)) + + if __name__ == '__main__': unittest.main()