Add a new timeout option to CLI

changelog.md

@@ -6,6 +6,7 @@
 This version is not released yet and is under active development.
 ```
 
+- \[mpm\] Add a `-t`/`--timeout` option to set the maximum duration of each CLI call. Defaults to 10 minutes.
 - \[mpm\] Drop support of Python 3.7.
 - \[scoop\] Fix parsing of Scoop version.
 - \[mpm\] Group platforms by family in the `managers` subcommand.

meta_package_manager/base.py

@@ -27,8 +27,9 @@
 from functools import cached_property
 from pathlib import Path
 from textwrap import dedent, indent, shorten
-from typing import ContextManager, Generator, Iterable, Iterator
+from typing import ContextManager, Generator, Iterable, Iterator, cast
 from unittest.mock import patch
+import subprocess
 
 from boltons.iterutils import flatten, unique
 from boltons.strutils import strip_ansi
@@ -41,7 +42,7 @@
     NestedArgs,
     args_cleanup,
     format_cli_prompt,
-    run_cmd,
+    env_copy,
 )
 
 from .version import TokenizedString, parse_version
@@ -330,6 +331,9 @@ class PackageManager(metaclass=MetaPackageManager):
     dry_run: bool = False
     """Do not actually perform any action, just simulate CLI calls."""
 
+    timeout: int | None = None
+    """Maximum number of seconds to wait for a CLI call to complete."""
+
     cli_errors: list[CLIError]
     """Accumulate all CLI errors encountered by the package manager."""
 
@@ -623,11 +627,6 @@ def run(self, *args: Arg | NestedArgs, extra_env: EnvVars | None = None) -> str:
           * returning ready-to-use normalized strings (dedented and stripped)
           * letting :option:`mpm --dry-run` and :option:`mpm --stop-on-error` have
             expected effect on execution
-
-        .. todo::
-
-            Move :option:`mpm --dry-run` option and this method to `click-extra
-            <https://github.com/kdeldycke/click-extra>`_.
         """
         # Casting to string helps serialize Path and Version objects.
         clean_args = args_cleanup(*args)
@@ -641,11 +640,16 @@ def run(self, *args: Arg | NestedArgs, extra_env: EnvVars | None = None) -> str:
             logging.warning(f"Dry-run: {cli_msg}")
         else:
             logging.debug(cli_msg)
-            code, output, error = run_cmd(
-                *clean_args,
-                extra_env=extra_env,
-                print_output=False,
+            result = subprocess.run(
+                clean_args,
+                capture_output=True,
+                timeout=self.timeout,
+                encoding="utf-8",
+                env=cast("subprocess._ENV", env_copy(extra_env)),
             )
+            code = result.returncode
+            output = result.stdout
+            error = result.stderr
 
         # Normalize messages.
         if error:

meta_package_manager/cli.py

@@ -34,6 +34,7 @@
     Choice,
     Context,
     File,
+    IntRange,
     Parameter,
     Section,
     argument,
@@ -224,6 +225,13 @@ def bar_plugin_path(ctx, param, value):
         default=False,
         help="Do not actually perform any action, just simulate CLI calls.",
     ),
+    option(
+        "-t",
+        "--timeout",
+        type=IntRange(min=0),
+        default=300,
+        help="Set maximum duration in seconds for each CLI call.",
+    ),
 )
 @option_group(
     "Output options",
@@ -275,6 +283,7 @@ def mpm(
     ignore_auto_updates,
     stop_on_error,
     dry_run,
+    timeout,
     description,
     sort_by,
     stats,
@@ -332,6 +341,7 @@ def remove_logging_override():
         # Does the manager should raise on error or not.
         stop_on_error=stop_on_error,
         dry_run=dry_run,
+        timeout=timeout,
     )
 
     # Load up current and new global options to the context for subcommand consumption.

meta_package_manager/pool.py

@@ -98,7 +98,7 @@ class ManagerPool:
     """A dict-like register, instanciating all supported package managers."""
 
     ALLOWED_EXTRA_OPTION: Final = frozenset(
-        {"ignore_auto_updates", "stop_on_error", "dry_run"},
+        {"ignore_auto_updates", "stop_on_error", "dry_run", "timeout"},
     )
     """List of extra options that are allowed to be set on managers during the use of
     the :py:func:`meta_package_manager.pool.ManagerPool.select_managers` helper
@@ -185,7 +185,7 @@ def select_managers(
         keep_unsupported: bool = False,
         drop_inactive: bool = True,
         implements_operation: Operations | None = None,
-        **extra_options: bool,
+        **extra_options: bool | int,
     ) -> Iterator[PackageManager]:
         """Utility method to extract a subset of the manager pool based on selection
         list (``keep`` parameter) and exclusion list (``drop`` parameter) criterion.

meta_package_manager/tests/test_cli.py

@@ -127,7 +127,7 @@ def check_manager_selection(
 
 
 class TestCommonCLI:
-    """Tests CLI behavior shared by all subcommands.
+    """Single tests for CLI behavior shared by all subcommands.
 
     If we have to, we only run the test on a single, non-destructive subcommand
     (like ``mpm installed`` or ``mpm managers``). Not all subcommands are tested.
@@ -156,6 +156,14 @@ def test_executable_module(self):
             expected_output = strip_ansi(expected_output)
         assert process.stdout == expected_output
 
+    def test_timeout(self, invoke):
+        """Check that the CLI exits with an error when a timeout is reached."""
+        result = invoke("--timeout", "1", "outdated")
+        assert result.exit_code == 1
+        assert not result.stdout
+        assert not result.stderr
+        assert isinstance(result.exception, subprocess.TimeoutExpired)
+
     @pytest.mark.parametrize(
         ("stats_arg", "active_stats"),
         (("--stats", True), ("--no-stats", False), (None, True)),