Add --executor CLI option to override the executor at runtime.

Also:
- Tweak output style to be more similar to poetry/cleo
- Add debug output when creating an executor
- Streamline executor creation logic to be simpler

docs/global_options.rst

@@ -107,14 +107,18 @@ The default behaviour is **auto**.
 
 For example, the following configuration will cause poe to ignore the poetry environment
 (if present), and instead use the virtualenv at the given location relative to the
-parent directory.
+parent directory. If no location is specified for a virtualenv then the default behavior is to use the virtualenv from ``./venv`` or ``./.venv`` if available.
 
 .. code-block:: toml
 
   [tool.poe.executor]
   type = "virtualenv"
   location = "myvenv"
 
+.. important::
+
+  This global option can be overridden at runtime by providing the ``--executor`` cli option before the task name with the name of the executor type to use.
+
 Change the default shell interpreter
 ------------------------------------
 

poethepoet/context.py

@@ -51,10 +51,6 @@ def __init__(
                 config_working_dir=config_part.cwd,
             )
 
-    @property
-    def executor_type(self) -> Optional[str]:
-        return self.config.executor["type"]
-
     def _get_dep_values(
         self, used_task_invocations: Mapping[str, Tuple[str, ...]]
     ) -> Dict[str, str]:
@@ -99,12 +95,18 @@ def get_executor(
     ) -> "PoeExecutor":
         from .executor import PoeExecutor
 
+        if not executor_config:
+            if self.ui["executor"]:
+                executor_config = {"type": self.ui["executor"]}
+            else:
+                executor_config = self.config.executor
+
         return PoeExecutor.get(
             invocation=invocation,
             context=self,
+            executor_config=executor_config,
             env=env,
             working_dir=working_dir,
-            executor_config=executor_config,
             capture_stdout=capture_stdout,
             dry=self.dry,
         )

poethepoet/executor/base.py

@@ -25,6 +25,8 @@
 
 # TODO: maybe invert the control so the executor is given a task to run?
 
+POE_DEBUG = os.environ.get("POE_DEBUG", "0") == "1"
+
 
 class MetaPoeExecutor(type):
     """
@@ -76,6 +78,9 @@ def __init__(
         self.dry = dry
         self._is_windows = sys.platform == "win32"
 
+        if POE_DEBUG:
+            print(f" . Initalizing {self.__class__.__name__}")
+
     @classmethod
     def works_with_context(cls, context: "RunContext") -> bool:
         return True
@@ -85,38 +90,28 @@ def get(
         cls,
         invocation: Tuple[str, ...],
         context: "RunContext",
+        executor_config: Mapping[str, str],
         env: "EnvVarsManager",
         working_dir: Optional[Path] = None,
-        executor_config: Optional[Mapping[str, str]] = None,
         capture_stdout: Union[str, bool] = False,
         dry: bool = False,
     ) -> "PoeExecutor":
-        """"""
-        # use task specific executor config or fallback to global
-        options = executor_config or context.config.executor
-        return cls._resolve_implementation(context, executor_config)(
-            invocation, context, options, env, working_dir, capture_stdout, dry
+        """
+        Create an executor.
+        """
+        return cls._resolve_implementation(context, executor_config["type"])(
+            invocation, context, executor_config, env, working_dir, capture_stdout, dry
         )
 
     @classmethod
-    def _resolve_implementation(
-        cls, context: "RunContext", executor_config: Optional[Mapping[str, str]]
-    ):
+    def _resolve_implementation(cls, context: "RunContext", executor_type: str):
         """
         Resolve to an executor class, either as specified in the available config or
         by making some reasonable assumptions based on visible features of the
         environment
         """
 
-        config_executor_type = context.executor_type
-        if executor_config:
-            executor_type = executor_config["type"]
-            if executor_type not in cls.__executor_types:
-                raise PoeException(
-                    f"Cannot instantiate unknown executor {executor_type!r}"
-                )
-            return cls.__executor_types[executor_type]
-        elif config_executor_type == "auto":
+        if executor_type == "auto":
             for impl in [
                 cls.__executor_types["poetry"],
                 cls.__executor_types["virtualenv"],
@@ -126,12 +121,13 @@ def _resolve_implementation(
 
             # Fallback to not using any particular environment
             return cls.__executor_types["simple"]
+
         else:
-            if config_executor_type not in cls.__executor_types:
+            if executor_type not in cls.__executor_types:
                 raise PoeException(
-                    "Cannot instantiate unknown executor" + repr(config_executor_type)
+                    f"Cannot instantiate unknown executor {executor_type!r}"
                 )
-            return cls.__executor_types[config_executor_type]
+            return cls.__executor_types[executor_type]
 
     def execute(
         self, cmd: Sequence[str], input: Optional[bytes] = None, use_exec: bool = False

poethepoet/ui.py

@@ -123,6 +123,16 @@ def build_parser(self) -> "ArgumentParser":
             help="Specify where to find the pyproject.toml",
         )
 
+        parser.add_argument(
+            "-e",
+            "--executor",
+            dest="executor",
+            metavar="EXECUTOR",
+            type=str,
+            default="",
+            help="Override the default task executor",
+        )
+
         # legacy --root parameter, keep for backwards compatibility but help output is
         # suppressed
         parser.add_argument(
@@ -216,9 +226,9 @@ def print_help(
         if verbosity >= 0:
             result.append(
                 (
-                    "<h2>USAGE</h2>",
+                    "<h2>Usage:</h2>",
                     f"  <u>{self.program_name}</u>"
-                    " [-h] [-v | -q] [-C PATH] [--ansi | --no-ansi]"
+                    " [global options]"
                     " task [task arguments]",
                 )
             )
@@ -230,7 +240,7 @@ def print_help(
             formatter.add_arguments(action_group._group_actions)
             formatter.end_section()
             result.append(
-                ("<h2>GLOBAL OPTIONS</h2>", *formatter.format_help().split("\n")[1:])
+                ("<h2>Global options:</h2>", *formatter.format_help().split("\n")[1:])
             )
 
             if tasks:
@@ -248,9 +258,9 @@ def print_help(
                     )
                     for task, (_, args) in tasks.items()
                 )
-                col_width = max(13, min(30, max_task_len))
+                col_width = max(20, min(30, max_task_len))
 
-                tasks_section = ["<h2>CONFIGURED TASKS</h2>"]
+                tasks_section = ["<h2>Configured tasks:</h2>"]
                 for task, (help_text, args_help) in tasks.items():
                     if task.startswith("_"):
                         continue

tests/test_api.py

@@ -3,7 +3,7 @@
 
 def test_customize_program_name(run_poe, projects):
     result = run_poe(program_name="boop")
-    assert "USAGE\n  boop [-h]" in result.capture
+    assert "Usage:\n  boop [global options] task" in result.capture
     assert result.stdout == ""
     assert result.stderr == ""
 

tests/test_cli.py

@@ -12,7 +12,7 @@ def test_call_no_args(run_poe):
     assert (
         "\nResult: No task specified.\n" in result.capture
     ), "Output should include status message"
-    assert "CONFIGURED TASKS\n  echo" in result.capture, "echo task should be in help"
+    assert "Configured tasks:\n  echo" in result.capture, "echo task should be in help"
 
 
 def test_call_with_directory(run_poe, projects):
@@ -25,8 +25,8 @@ def test_call_with_directory(run_poe, projects):
         "\nResult: No task specified.\n" in result.capture
     ), "Output should include status message"
     assert (
-        "CONFIGURED TASKS\n"
-        "  echo                 It says what you say" in result.capture
+        "Configured tasks:\n"
+        "  echo                  It says what you say" in result.capture
     ), "echo task should be in help"
 
 
@@ -40,8 +40,8 @@ def test_call_with_directory_set_via_env(run_poe_subproc, projects):
         "\nResult: No task specified.\n" in result.capture
     ), "Output should include status message"
     assert (
-        "CONFIGURED TASKS\n"
-        "  echo                 It says what you say" in result.capture
+        "Configured tasks:\n"
+        "  echo                  It says what you say" in result.capture
     ), "echo task should be in help"
 
 
@@ -56,8 +56,8 @@ def test_call_with_root(run_poe, projects):
         "\nResult: No task specified.\n" in result.capture
     ), "Output should include status message"
     assert (
-        "CONFIGURED TASKS\n"
-        "  echo                 It says what you say" in result.capture
+        "Configured tasks:\n"
+        "  echo                  It says what you say" in result.capture
     ), "echo task should be in help"
 
 
@@ -111,7 +111,7 @@ def test_documentation_of_task_named_args(run_poe):
     ), "Output should include status message"
 
     assert re.search(
-        r"CONFIGURED TASKS\n"
+        r"Configured tasks:\n"
         r"  composite_task          \s+\n"
         r"  echo-args               \s+\n"
         r"  static-args-test        \s+\n"

tests/test_executors.py

@@ -117,3 +117,37 @@ def test_simple_executor(run_poe_subproc):
         f"/tests/fixtures/simple_project/venv/lib/python{PY_V}/site-packages/poe_test_package/__init__.py\n"
     )
     assert result.stderr == ""
+
+
+def test_override_executor(run_poe_subproc, with_virtualenv_and_venv, projects):
+    """
+    This test includes two scenarios
+
+    1. A variation on test_virtualenv_executor_fails_without_venv_dir except that
+       because we force use of the simple executor we don't get the error
+
+    2. A variation on test_virtualenv_executor_activates_venv except that because we
+       force use of the simple executor we don't get the virtual_env
+    """
+
+    # 1.
+    venv_path = projects["venv"].joinpath("myvenv")
+    assert (
+        not venv_path.is_dir()
+    ), f"This test requires the virtualenv not to already exist at {venv_path}!"
+    result = run_poe_subproc("--executor", "simple", "show-env", project="venv")
+    assert (
+        f"Error: Could not find valid virtualenv at configured location: {venv_path}"
+        not in result.capture
+    )
+    assert result.stderr == ""
+
+    # 2.
+    venv_path = projects["venv"].joinpath("myvenv")
+    for _ in with_virtualenv_and_venv(
+        venv_path, ["./tests/fixtures/packages/poe_test_helpers"]
+    ):
+        result = run_poe_subproc("-e", "simple", "show-env", project="venv")
+        assert result.capture == "Poe => poe_test_env\n"
+        assert f"VIRTUAL_ENV={venv_path}" not in result.stdout
+        assert result.stderr == ""

tests/test_includes.py

@@ -20,11 +20,11 @@ def _init_git_submodule(projects):
 def test_docs_for_include_toml_file(run_poe_subproc):
     result = run_poe_subproc(project="includes")
     assert (
-        "CONFIGURED TASKS\n"
-        "  echo           says what you say\n"
-        "  greet          \n"
-        "  greet1         \n"
-        "  greet2         Issue a greeting from the Iberian Peninsula\n"
+        "Configured tasks:\n"
+        "  echo                  says what you say\n"
+        "  greet                 \n"
+        "  greet1                \n"
+        "  greet2                Issue a greeting from the Iberian Peninsula\n"
     ) in result.capture
     assert result.stdout == ""
     assert result.stderr == ""
@@ -49,7 +49,7 @@ def test_docs_for_multiple_includes(run_poe_subproc, projects):
         f'-C={projects["includes/multiple_includes"]}',
     )
     assert (
-        "CONFIGURED TASKS\n"
+        "Configured tasks:\n"
         "  echo                    says what you say\n"
         "  greet                   \n"
         "  greet1                  \n"
@@ -102,11 +102,11 @@ def test_docs_for_only_includes(run_poe_subproc, projects):
         f'-C={projects["includes/only_includes"]}',
     )
     assert (
-        "CONFIGURED TASKS\n"
-        "  echo           This is ignored because it's already defined!\n"  # or not
-        "  greet          \n"
-        "  greet1         \n"
-        "  greet2         Issue a greeting from the Iberian Peninsula\n"
+        "Configured tasks:\n"
+        "  echo                  This is ignored because it's already defined!\n"
+        "  greet                 \n"
+        "  greet1                \n"
+        "  greet2                Issue a greeting from the Iberian Peninsula\n"
     ) in result.capture
     assert result.stdout == ""
     assert result.stderr == ""
@@ -115,14 +115,14 @@ def test_docs_for_only_includes(run_poe_subproc, projects):
 def test_monorepo_contains_only_expected_tasks(run_poe_subproc, projects):
     result = run_poe_subproc(project="monorepo")
     assert result.capture.endswith(
-        "CONFIGURED TASKS\n"
-        "  get_cwd_0      \n"
-        "  get_cwd_1      \n"
-        "  add            \n"
-        "  get_cwd_2      \n"
-        "  subproj3_env   \n"
-        "  get_cwd_3      \n"
-        "  subproj4_env   \n\n\n"
+        "Configured tasks:\n"
+        "  get_cwd_0             \n"
+        "  get_cwd_1             \n"
+        "  add                   \n"
+        "  get_cwd_2             \n"
+        "  subproj3_env          \n"
+        "  get_cwd_3             \n"
+        "  subproj4_env          \n\n\n"
     )
     assert result.stdout == ""
     assert result.stderr == ""
@@ -131,11 +131,11 @@ def test_monorepo_contains_only_expected_tasks(run_poe_subproc, projects):
 def test_monorepo_can_also_include_parent(run_poe_subproc, projects, is_windows):
     result = run_poe_subproc(cwd=projects["monorepo/subproject_2"])
     assert result.capture.endswith(
-        "CONFIGURED TASKS\n"
-        "  add            \n"
-        "  get_cwd_2      \n"
-        "  extra_task     \n"
-        "  get_cwd_0      \n\n\n"
+        "Configured tasks:\n"
+        "  add                   \n"
+        "  get_cwd_2             \n"
+        "  extra_task            \n"
+        "  get_cwd_0             \n\n\n"
     )
     assert result.stdout == ""
     assert result.stderr == ""

tests/test_poetry_plugin.py

@@ -145,7 +145,7 @@ def test_running_tasks_with_poe_command_prefix_missing_args(run_poetry, projects
         ["foo"],
         cwd=projects["poetry_plugin/with_prefix"].parent,
     )
-    assert "USAGE\n  poetry foo [-h]" in result.stdout
+    assert "Usage:\n  poetry foo [global options]" in result.stdout
     # assert result.stderr == ""