diff --git a/src/mcpm/commands/install.py b/src/mcpm/commands/install.py index 80c4b17..b45356f 100644 --- a/src/mcpm/commands/install.py +++ b/src/mcpm/commands/install.py @@ -20,6 +20,7 @@ from mcpm.profile.profile_config import ProfileConfigManager from mcpm.schemas.full_server_config import FullServerConfig from mcpm.utils.config import NODE_EXECUTABLES, ConfigManager +from mcpm.utils.non_interactive import is_explicit_non_interactive, should_force_operation from mcpm.utils.repository import RepositoryManager from mcpm.utils.rich_click_config import click @@ -88,7 +89,7 @@ def _replace_node_executable(server_config: ServerConfig) -> ServerConfig: def global_add_server(server_config: ServerConfig, force: bool = False) -> bool: """Add a server to the global MCPM configuration.""" if global_config_manager.server_exists(server_config.name) and not force: - console.print(f"[bold red]Error:[/] Server '{server_config.name}' already exists in global configuration.") + console.print(f"[bold red]Error:[/ ] Server '{server_config.name}' already exists in global configuration.") console.print("Use --force to override.") return False @@ -96,18 +97,36 @@ def global_add_server(server_config: ServerConfig, force: bool = False) -> bool: return global_config_manager.add_server(server_config, force) -def prompt_with_default(prompt_text, default="", hide_input=False, required=False): +def prompt_with_default(prompt_text, default="", hide_input=False, required=False, force=False): """Prompt the user with a default value that can be edited directly. + In non-interactive mode (via --force or MCPM_NON_INTERACTIVE), this function + returns the default value immediately. If a required value has no default + in non-interactive mode, it raises click.UsageError. + Args: prompt_text: The prompt text to display default: The default value to show in the prompt hide_input: Whether to hide the input (for passwords) required: Whether this is a required field + force: Whether to force non-interactive mode Returns: The user's input or the default value if empty """ + # Check for explicit non-interactive mode (Env Var) or Force flag + # We use is_explicit_non_interactive() instead of the broader is_non_interactive() + # because the latter includes isatty() checks. In test environments using CliRunner, + # isatty() returns True, which would incorrectly skip our mocked prompts if we checked it here. + # We specifically want to allow interaction in tests unless the Env Var is set. + if is_explicit_non_interactive() or should_force_operation(force): + if default: + return default + if required: + # Cannot fulfill required argument without default in non-interactive mode + raise click.UsageError("A required value has no default and cannot be prompted in non-interactive mode.") + return "" + # if default: # console.print(f"Default: [yellow]{default}[/]") @@ -142,7 +161,7 @@ def prompt_with_default(prompt_text, default="", hide_input=False, required=Fals # Empty result for required field without default is not allowed if not result.strip() and required and not default: console.print("[yellow]Warning: Required value cannot be empty.[/]") - return prompt_with_default(prompt_text, default, hide_input, required) + return prompt_with_default(prompt_text, default, hide_input, required, force) return result except (KeyboardInterrupt, EOFError): @@ -171,12 +190,13 @@ def install(server_name, force=False, alias=None): config_name = alias or server_name # All servers are installed to global configuration - console.print("[yellow]Installing server to global configuration...[/]") + console_stderr = Console(stderr=True) + console_stderr.print("[yellow]Installing server to global configuration...[/]") # Get server metadata from repository server_metadata = repo_manager.get_server_metadata(server_name) if not server_metadata: - console.print(f"[bold red]Error:[/] Server '{server_name}' not found in registry.") + console.print(f"[bold red]Error:[/ ] Server '{server_name}' not found in registry.") console.print(f"Available servers: {', '.join(repo_manager._fetch_servers().keys())}") return @@ -195,7 +215,10 @@ def install(server_name, force=False, alias=None): # Confirm addition alias_text = f" as '{alias}'" if alias else "" - if not force and not Confirm.ask(f"Install this server to global configuration{alias_text}?"): + # Bypass confirmation if force flag is set OR explicit non-interactive mode is enabled + if not (should_force_operation(force) or is_explicit_non_interactive()) and not Confirm.ask( + f"Install this server to global configuration{alias_text}?" + ): console.print("[yellow]Operation cancelled.[/]") return @@ -239,8 +262,8 @@ def install(server_name, force=False, alias=None): method_id = next(iter(installations)) selected_method = installations[method_id] - # If multiple methods are available and not forced, offer selection - if len(installations) > 1 and not force: + # If multiple methods are available and not forced/non-interactive, offer selection + if len(installations) > 1 and not (should_force_operation(force) or is_explicit_non_interactive()): console.print("\n[bold]Available installation methods:[/]") methods_list = [] @@ -349,6 +372,7 @@ def install(server_name, force=False, alias=None): default=env_value, hide_input=_should_hide_input(arg_name), required=is_required, + force=force, ) if user_value != env_value: # User provided a different value @@ -366,6 +390,7 @@ def install(server_name, force=False, alias=None): default=example if example else "", hide_input=_should_hide_input(arg_name), required=is_required, + force=force, ) # Only add non-empty values to the environment @@ -419,7 +444,7 @@ def install(server_name, force=False, alias=None): if has_non_standard_argument_define: # no matter in argument / env console.print( - "[bold yellow]WARNING:[/] [bold]Non-standard argument format detected in server configuration.[/]\n" + "[bold yellow]WARNING:[/ ] [bold]Non-standard argument format detected in server configuration.[/]\n" "[bold cyan]Future versions of MCPM will standardize all arguments in server configuration to use ${VARIABLE_NAME} format.[/]\n" "[bold]Please verify that your input arguments are correctly recognized.[/]\n" ) @@ -460,7 +485,7 @@ def install(server_name, force=False, alias=None): ) # Add server to global configuration - success = global_add_server(full_server_config.to_server_config(), force) + success = global_add_server(full_server_config.to_server_config(), should_force_operation(force)) if success: # Server has been successfully added to the global configuration @@ -523,7 +548,7 @@ def extract_from_value(value): # Check all fields in the installation method for key, value in installation_method.items(): - if key not in ["type", "description", "recommended"]: # Skip metadata fields + if key not in ["type", "description", "recommended"]: extract_from_value(value) return referenced diff --git a/src/mcpm/commands/uninstall.py b/src/mcpm/commands/uninstall.py index 4e058f4..c4dc978 100644 --- a/src/mcpm/commands/uninstall.py +++ b/src/mcpm/commands/uninstall.py @@ -7,6 +7,7 @@ from mcpm.global_config import GlobalConfigManager from mcpm.utils.display import print_server_config +from mcpm.utils.non_interactive import is_explicit_non_interactive, should_force_operation from mcpm.utils.rich_click_config import click console = Console() @@ -17,14 +18,14 @@ def global_get_server(server_name: str): """Get a server from the global MCPM configuration.""" server = global_config_manager.get_server(server_name) if not server: - console.print(f"[bold red]Error:[/] Server '{server_name}' not found in global configuration.") + console.print(f"[bold red]Error:[/ ] Server '{server_name}' not found in global configuration.") return server def global_remove_server(server_name: str) -> bool: """Remove a server from the global MCPM configuration and clean up profile tags.""" if not global_config_manager.server_exists(server_name): - console.print(f"[bold red]Error:[/] Server '{server_name}' not found in global configuration.") + console.print(f"[bold red]Error:[/ ] Server '{server_name}' not found in global configuration.") return False # Remove from global config (this automatically removes all profile tags) @@ -54,13 +55,13 @@ def uninstall(server_name, force): return # Error message already printed by global_get_server # Display server information before removal - console.print(f"\n[bold cyan]Server information for:[/] {server_name}") + console.print(f"\n[bold cyan]Server information for:[/ ] {server_name}") print_server_config(server_info) - # Get confirmation if --force is not used - if not force: - console.print(f"\n[bold yellow]Are you sure you want to remove:[/] {server_name}") + # Get confirmation if --force is not used and not in non-interactive mode + if not (should_force_operation(force) or is_explicit_non_interactive()): + console.print(f"\n[bold yellow]Are you sure you want to remove:[/ ] {server_name}") console.print("[italic]To bypass this confirmation, use --force[/]") # Use Rich's Confirm for a better user experience confirmed = Confirm.ask("Proceed with removal?") @@ -69,13 +70,13 @@ def uninstall(server_name, force): return # Log the removal action - console.print(f"[bold red]Removing MCP server from global configuration:[/] {server_name}") + console.print(f"[bold red]Removing MCP server from global configuration:[/ ] {server_name}") # Remove from global configuration success = global_remove_server(server_name) if success: - console.print(f"[green]Successfully removed server:[/] {server_name}") + console.print(f"[green]Successfully removed server:[/ ] {server_name}") console.print("[dim]Note: Server has been removed from global config. Profile tags are also cleared.[/]") else: - console.print(f"[bold red]Error:[/] Failed to remove server '{server_name}'.") + console.print(f"[bold red]Error:[/ ] Failed to remove server '{server_name}'.") \ No newline at end of file diff --git a/src/mcpm/utils/non_interactive.py b/src/mcpm/utils/non_interactive.py index c2d4efd..28f4ea9 100644 --- a/src/mcpm/utils/non_interactive.py +++ b/src/mcpm/utils/non_interactive.py @@ -7,6 +7,19 @@ from typing import Dict, List, Optional +def is_explicit_non_interactive() -> bool: + """ + Check if non-interactive mode is explicitly enabled via environment variable. + + This excludes implicit detection (like isatty) to avoid issues in tests or + environments where TTY detection behaves unexpectedly but automation is not desired. + + Returns: + True if MCPM_NON_INTERACTIVE environment variable is set to 'true' + """ + return os.getenv("MCPM_NON_INTERACTIVE", "").lower() == "true" + + def is_non_interactive() -> bool: """ Check if running in non-interactive mode. @@ -17,7 +30,7 @@ def is_non_interactive() -> bool: - Running in a CI environment """ # Check explicit non-interactive flag - if os.getenv("MCPM_NON_INTERACTIVE", "").lower() == "true": + if is_explicit_non_interactive(): return True # Check if not connected to a TTY @@ -32,13 +45,17 @@ def is_non_interactive() -> bool: return False -def should_force_operation() -> bool: +def should_force_operation(cli_force_flag: bool = False) -> bool: """ Check if operations should be forced (skip confirmations). - Returns True if MCPM_FORCE environment variable is set to 'true'. + Args: + cli_force_flag: Boolean flag from CLI args (e.g. --force) + + Returns: + True if cli_force_flag is True OR MCPM_FORCE environment variable is set to 'true'. """ - return os.getenv("MCPM_FORCE", "").lower() == "true" + return cli_force_flag or os.getenv("MCPM_FORCE", "").lower() == "true" def should_output_json() -> bool: @@ -304,4 +321,4 @@ def merge_server_config_updates( else: updated_config["headers"] = new_headers - return updated_config + return updated_config \ No newline at end of file diff --git a/tests/test_add.py b/tests/test_add.py index aa30974..89506d2 100644 --- a/tests/test_add.py +++ b/tests/test_add.py @@ -41,10 +41,16 @@ def test_add_server(windsurf_manager, monkeypatch, tmp_path): ) # Mock prompt_toolkit's prompt to return our test values - with patch("prompt_toolkit.PromptSession.prompt", side_effect=["json", "test-api-key"]): + # Note: With --force, the CLI will look for env vars for required args instead of prompting + monkeypatch.setenv("fmt", "json") + monkeypatch.setenv("API_KEY", "test-api-key") + + # We still patch PromptSession to ensure it's NOT called (or ignored if called incorrectly) + with patch("prompt_toolkit.PromptSession.prompt") as mock_prompt: runner = CliRunner() result = runner.invoke(install, ["server-test", "--force", "--alias", "test"]) assert result.exit_code == 0 + mock_prompt.assert_not_called() # Check that the server was added to global configuration with alias server = global_config_manager.get_server("test") @@ -95,8 +101,13 @@ def test_add_server_with_missing_arg(windsurf_manager, monkeypatch, tmp_path): # Instead of mocking Console and Progress, we'll mock key methods directly # This is a simpler approach that avoids complex mock setup + + # Set environment variables for required arguments + monkeypatch.setenv("fmt", "json") + monkeypatch.setenv("API_KEY", "test-api-key") + with ( - patch("prompt_toolkit.PromptSession.prompt", side_effect=["json", "test-api-key"]), + patch("prompt_toolkit.PromptSession.prompt") as mock_prompt, patch("rich.progress.Progress.start"), patch("rich.progress.Progress.stop"), patch("rich.progress.Progress.add_task"), @@ -111,6 +122,7 @@ def test_add_server_with_missing_arg(windsurf_manager, monkeypatch, tmp_path): print(f"Output: {result.stdout}") assert result.exit_code == 0 + mock_prompt.assert_not_called() # Check that the server was added with alias and the missing argument is replaced with empty string server = global_config_manager.get_server("test-missing-arg") @@ -166,8 +178,12 @@ def test_add_server_with_empty_args(windsurf_manager, monkeypatch, tmp_path): ) # Mock prompt responses for required arguments only + monkeypatch.setenv("fmt", "json") + monkeypatch.setenv("API_KEY", "test-api-key") + # OPTIONAL is not set, simulating empty/missing optional arg + with ( - patch("prompt_toolkit.PromptSession.prompt", side_effect=["json", "test-api-key"]), + patch("prompt_toolkit.PromptSession.prompt") as mock_prompt, patch("rich.progress.Progress.start"), patch("rich.progress.Progress.stop"), patch("rich.progress.Progress.add_task"), @@ -176,6 +192,7 @@ def test_add_server_with_empty_args(windsurf_manager, monkeypatch, tmp_path): result = runner.invoke(install, ["server-test", "--force", "--alias", "test-empty-args"]) assert result.exit_code == 0 + mock_prompt.assert_not_called() # Check that the server was added and optional arguments are empty server = global_config_manager.get_server("test-empty-args") @@ -269,14 +286,18 @@ def test_add_server_with_configured_npx(windsurf_manager, monkeypatch, tmp_path) ) # Mock Rich's progress display to prevent 'Only one live display may be active at once' error + monkeypatch.setenv("fmt", "json") + monkeypatch.setenv("API_KEY", "test-api-key") + with ( patch("rich.progress.Progress.__enter__", return_value=Mock()), patch("rich.progress.Progress.__exit__"), - patch("prompt_toolkit.PromptSession.prompt", side_effect=["json", "test-api-key"]), + patch("prompt_toolkit.PromptSession.prompt") as mock_prompt, ): runner = CliRunner() result = runner.invoke(install, ["server-test", "--force", "--alias", "test"]) assert result.exit_code == 0 + mock_prompt.assert_not_called() # Check that the server was added with alias server = global_config_manager.get_server("test") @@ -384,14 +405,11 @@ def test_add_server_with_filtered_arguments(windsurf_manager, monkeypatch, tmp_p # Mock prompt_toolkit's prompt to return our test values # Should only be called once for API_KEY since that's the only referenced variable - prompt_calls = [] - - def mock_prompt_func(*args, **kwargs): - prompt_calls.append(kwargs.get("message", "")) - return "test-api-key" - + # With force=True, we use env vars + monkeypatch.setenv("API_KEY", "test-api-key") + with ( - patch("prompt_toolkit.PromptSession.prompt", side_effect=mock_prompt_func), + patch("prompt_toolkit.PromptSession.prompt") as mock_prompt, patch("rich.progress.Progress.start"), patch("rich.progress.Progress.stop"), patch("rich.progress.Progress.add_task"), @@ -399,12 +417,7 @@ def mock_prompt_func(*args, **kwargs): runner = CliRunner() result = runner.invoke(install, ["test-server", "--force"]) assert result.exit_code == 0 - - # Check that only API_KEY was prompted for - assert len(prompt_calls) == 1 - assert "API_KEY" in str(prompt_calls[0]) - assert "DATABASE_URL" not in str(prompt_calls[0]) - assert "UNUSED_VAR" not in str(prompt_calls[0]) + mock_prompt.assert_not_called() # Check that the server was added correctly server = global_config_manager.get_server("test-server") @@ -449,8 +462,10 @@ def test_add_http_server_with_headers(windsurf_manager, monkeypatch, tmp_path): ) # Mock prompt_toolkit's prompt to return our test values + monkeypatch.setenv("API_TOKEN", "test-token-123") + with ( - patch("prompt_toolkit.PromptSession.prompt", side_effect=["test-token-123"]), + patch("prompt_toolkit.PromptSession.prompt") as mock_prompt, patch("rich.progress.Progress.start"), patch("rich.progress.Progress.stop"), patch("rich.progress.Progress.add_task"), @@ -458,6 +473,7 @@ def test_add_http_server_with_headers(windsurf_manager, monkeypatch, tmp_path): runner = CliRunner() result = runner.invoke(install, ["api-server", "--force"]) assert result.exit_code == 0 + mock_prompt.assert_not_called() # Check that the server was added to global configuration as RemoteServerConfig server = global_config_manager.get_server("api-server") @@ -465,4 +481,4 @@ def test_add_http_server_with_headers(windsurf_manager, monkeypatch, tmp_path): assert isinstance(server, RemoteServerConfig) assert server.url == "https://api.example.com/mcp" # Check headers were properly set with variable replacement - assert server.headers == {"Authorization": "Bearer test-token-123", "X-API-Version": "1.0"} + assert server.headers == {"Authorization": "Bearer test-token-123", "X-API-Version": "1.0"} \ No newline at end of file