fix(mcpm): Address PR #283 feedback - explicit non-interactive checks and test fixes

- Add is_explicit_non_interactive() helper to avoid TTY detection issues in tests

- Update install/uninstall logic to strictly respect force/non-interactive flags

- Pass force flag correctly to prompt_with_default

- Update prompt_with_default docstring

- Update tests to use env vars for input when --force is used

Author: JRedeker

src/mcpm/commands/install.py

@@ -20,7 +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 should_force_operation
+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
 
@@ -65,7 +65,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
 
@@ -76,6 +76,10 @@ def global_add_server(server_config: ServerConfig, force: bool = False) -> bool:
 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
@@ -86,11 +90,12 @@ def prompt_with_default(prompt_text, default="", hide_input=False, required=Fals
     Returns:
         The user's input or the default value if empty
     """
-    # Check for explicit non-interactive mode (Env Var)
-    # We do NOT check is_non_interactive() here because it includes isatty(),
-    # which returns True in tests (CliRunner), causing us to skip mocked prompts.
-    # Users desiring non-interactive behavior must set MCPM_NON_INTERACTIVE=true.
-    if os.getenv("MCPM_NON_INTERACTIVE", "").lower() == "true" or should_force_operation(force):
+    # 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:
@@ -124,7 +129,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:
@@ -159,7 +164,7 @@ def install(server_name, force=False, alias=None):
     # 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
 
@@ -178,7 +183,8 @@ def install(server_name, force=False, alias=None):
 
     # Confirm addition
     alias_text = f" as '{alias}'" if alias else ""
-    if not should_force_operation(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
 
@@ -222,8 +228,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 should_force_operation(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 = []
 
@@ -332,6 +338,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
@@ -349,6 +356,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
@@ -402,7 +410,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"
         )
@@ -506,7 +514,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
@@ -569,10 +577,12 @@ def _replace_argument_variables(value: str, prev_value: str, variables: dict) ->
     if matched:
         original, var_name = matched.group(0), matched.group(1)
         # Use empty string as default when key not found
-        return value.replace(original, variables.get(var_name, "")), ReplacementStatus.STANDARD_REPLACE
+        return value.replace(original, variables.get(var_name, "")),
+    ReplacementStatus.STANDARD_REPLACE
 
     # arg: --VAR=your var value
-    key_value_match = re.match(r"^([A-Z_]+)=(.+)$", value)
+    key_value_match = re.match(r"^([A-Z_]+)=(.+)$
+", value)
     if key_value_match:
         arg_key = key_value_match.group(1)
         if arg_key in variables:
@@ -591,4 +601,4 @@ def _replace_argument_variables(value: str, prev_value: str, variables: dict) ->
         return value, ReplacementStatus.NOT_REPLACED
 
     # nothing to replace
-    return value, ReplacementStatus.NOT_REPLACED
+    return value, ReplacementStatus.NOT_REPLACED

src/mcpm/commands/uninstall.py

@@ -7,7 +7,7 @@
 
 from mcpm.global_config import GlobalConfigManager
 from mcpm.utils.display import print_server_config
-from mcpm.utils.non_interactive import should_force_operation
+from mcpm.utils.non_interactive import is_explicit_non_interactive, should_force_operation
 from mcpm.utils.rich_click_config import click
 
 console = Console()
@@ -18,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)
@@ -55,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 should_force_operation(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?")
@@ -70,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}'.")

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
@@ -308,4 +321,4 @@ def merge_server_config_updates(
         else:
             updated_config["headers"] = new_headers
 
-    return updated_config
+    return updated_config

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,27 +405,19 @@ 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"),
     ):
         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,20 +462,23 @@ 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"),
     ):
         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")
     assert server is not None
     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"}