From bd5b4cef616c14bb93694bcb29d3abcf2af0692c Mon Sep 17 00:00:00 2001 From: Tom Duckering Date: Tue, 15 Apr 2025 13:33:51 +0100 Subject: [PATCH 1/6] First attempt at allowing list of strings for commands Why? Because it's safer to pass string array to the kernel as it's safer. --- invoke/runners.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/invoke/runners.py b/invoke/runners.py index f1c888f44..82edc34c2 100644 --- a/invoke/runners.py +++ b/invoke/runners.py @@ -122,7 +122,7 @@ def __init__(self, context: "Context") -> None: self._asynchronous = False self._disowned = False - def run(self, command: str, **kwargs: Any) -> Optional["Result"]: + def run(self, command: str | List[str], **kwargs: Any) -> Optional["Result"]: """ Execute ``command``, returning an instance of `Result` once complete. @@ -144,7 +144,7 @@ def run(self, command: str, **kwargs: Any) -> Optional["Result"]: the ``echo`` keyword, etc). The base default values are described in the parameter list below. - :param str command: The shell command to execute. + :param str | List[str] command: The shell command to execute. :param bool asynchronous: When set to ``True`` (default ``False``), enables asynchronous @@ -428,7 +428,7 @@ def _setup(self, command: str, kwargs: Any) -> None: encoding=self.encoding, ) - def _run_body(self, command: str, **kwargs: Any) -> Optional["Result"]: + def _run_body(self, command: str | List[str], **kwargs: Any) -> Optional["Result"]: # Prepare all the bits n bobs. self._setup(command, kwargs) # If dry-run, stop here. @@ -1043,7 +1043,7 @@ def process_is_finished(self) -> bool: """ raise NotImplementedError - def start(self, command: str, shell: str, env: Dict[str, Any]) -> None: + def start(self, command: str | List[str], shell: str, env: Dict[str, Any]) -> None: """ Initiate execution of ``command`` (via ``shell``, with ``env``). @@ -1305,7 +1305,7 @@ def close_proc_stdin(self) -> None: "Unable to close missing subprocess or stdin!" ) - def start(self, command: str, shell: str, env: Dict[str, Any]) -> None: + def start(self, command: str | List[str], shell: str, env: Dict[str, Any]) -> None: if self.using_pty: if pty is None: # Encountered ImportError err = "You indicated pty=True, but your platform doesn't support the 'pty' module!" # noqa From f0d7b2755331a402323ba389fba3d28103e3e70e Mon Sep 17 00:00:00 2001 From: Tom Duckering Date: Tue, 15 Apr 2025 13:57:36 +0100 Subject: [PATCH 2/6] Use Union syntax for the type hint --- invoke/runners.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/invoke/runners.py b/invoke/runners.py index 82edc34c2..107a8aebc 100644 --- a/invoke/runners.py +++ b/invoke/runners.py @@ -19,6 +19,7 @@ Optional, Tuple, Type, + Union, ) # Import some platform-specific things at top level so they can be mocked for @@ -122,7 +123,7 @@ def __init__(self, context: "Context") -> None: self._asynchronous = False self._disowned = False - def run(self, command: str | List[str], **kwargs: Any) -> Optional["Result"]: + def run(self, command: Union[str, List[str]], **kwargs: Any) -> Optional["Result"]: """ Execute ``command``, returning an instance of `Result` once complete. @@ -144,7 +145,7 @@ def run(self, command: str | List[str], **kwargs: Any) -> Optional["Result"]: the ``echo`` keyword, etc). The base default values are described in the parameter list below. - :param str | List[str] command: The shell command to execute. + :param Union[str, List[str]] command: The shell command to execute. :param bool asynchronous: When set to ``True`` (default ``False``), enables asynchronous @@ -400,7 +401,7 @@ def run(self, command: str | List[str], **kwargs: Any) -> Optional["Result"]: def echo(self, command: str) -> None: print(self.opts["echo_format"].format(command=command)) - def _setup(self, command: str, kwargs: Any) -> None: + def _setup(self, command: Union[str, List[str]], kwargs: Any) -> None: """ Prepare data on ``self`` so we're ready to start running. """ @@ -428,7 +429,7 @@ def _setup(self, command: str, kwargs: Any) -> None: encoding=self.encoding, ) - def _run_body(self, command: str | List[str], **kwargs: Any) -> Optional["Result"]: + def _run_body(self, command: Union[str, List[str]], **kwargs: Any) -> Optional["Result"]: # Prepare all the bits n bobs. self._setup(command, kwargs) # If dry-run, stop here. From f6cebe0cf584fd2d2b4304881e67393589d89d58 Mon Sep 17 00:00:00 2001 From: Tom Duckering Date: Tue, 15 Apr 2025 14:03:59 +0100 Subject: [PATCH 3/6] Fix echo function signature to accept union and handle list of string or string when printing --- invoke/runners.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/invoke/runners.py b/invoke/runners.py index 107a8aebc..ebb373b7e 100644 --- a/invoke/runners.py +++ b/invoke/runners.py @@ -398,8 +398,9 @@ def run(self, command: Union[str, List[str]], **kwargs: Any) -> Optional["Result if not (self._asynchronous or self._disowned): self.stop() - def echo(self, command: str) -> None: - print(self.opts["echo_format"].format(command=command)) + def echo(self, command: Union[str, List[str]]) -> None: + command_string = command if isinstance(command, str) else " ".join(command) + print(self.opts["echo_format"].format(command=command_string)) def _setup(self, command: Union[str, List[str]], kwargs: Any) -> None: """ From 15a35ae0f74cb394fe4971a6d0cf2ed07821e0db Mon Sep 17 00:00:00 2001 From: Tom Duckering Date: Tue, 15 Apr 2025 14:12:31 +0100 Subject: [PATCH 4/6] handle string or list of strings properly for execve --- invoke/runners.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/invoke/runners.py b/invoke/runners.py index ebb373b7e..a88b0d2d1 100644 --- a/invoke/runners.py +++ b/invoke/runners.py @@ -1334,7 +1334,8 @@ def start(self, command: str | List[str], shell: str, env: Dict[str, Any]) -> No # for now. # NOTE: stdlib subprocess (actually its posix flavor, which is # written in C) uses either execve or execv, depending. - os.execve(shell, [shell, "-c", command], env) + command_parts = [command] if isinstance(command, str) else command + os.execve(shell, [shell, "-c", *command_parts], env) else: self.process = Popen( command, From 9f00bdf382375da59992e8c89948fc22aa1ef91e Mon Sep 17 00:00:00 2001 From: Tom Duckering Date: Tue, 15 Apr 2025 14:15:40 +0100 Subject: [PATCH 5/6] Fix types to use older `Union` notation missed some in the earlier commit --- invoke/runners.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/invoke/runners.py b/invoke/runners.py index a88b0d2d1..4d12bf883 100644 --- a/invoke/runners.py +++ b/invoke/runners.py @@ -1045,7 +1045,7 @@ def process_is_finished(self) -> bool: """ raise NotImplementedError - def start(self, command: str | List[str], shell: str, env: Dict[str, Any]) -> None: + def start(self, command: Union[str, List[str]], shell: str, env: Dict[str, Any]) -> None: """ Initiate execution of ``command`` (via ``shell``, with ``env``). @@ -1307,7 +1307,7 @@ def close_proc_stdin(self) -> None: "Unable to close missing subprocess or stdin!" ) - def start(self, command: str | List[str], shell: str, env: Dict[str, Any]) -> None: + def start(self, command: Union[str, List[str]], shell: str, env: Dict[str, Any]) -> None: if self.using_pty: if pty is None: # Encountered ImportError err = "You indicated pty=True, but your platform doesn't support the 'pty' module!" # noqa From 8212ee9f86123d611a471ebb99a4f14169effaa7 Mon Sep 17 00:00:00 2001 From: Tom Duckering Date: Tue, 15 Apr 2025 14:18:07 +0100 Subject: [PATCH 6/6] Appease blacken formatter --- invoke/runners.py | 24 ++++++++++++++++++------ 1 file changed, 18 insertions(+), 6 deletions(-) diff --git a/invoke/runners.py b/invoke/runners.py index 4d12bf883..e2d21cdd5 100644 --- a/invoke/runners.py +++ b/invoke/runners.py @@ -123,7 +123,9 @@ def __init__(self, context: "Context") -> None: self._asynchronous = False self._disowned = False - def run(self, command: Union[str, List[str]], **kwargs: Any) -> Optional["Result"]: + def run( + self, command: Union[str, List[str]], **kwargs: Any + ) -> Optional["Result"]: """ Execute ``command``, returning an instance of `Result` once complete. @@ -399,7 +401,9 @@ def run(self, command: Union[str, List[str]], **kwargs: Any) -> Optional["Result self.stop() def echo(self, command: Union[str, List[str]]) -> None: - command_string = command if isinstance(command, str) else " ".join(command) + command_string = ( + command if isinstance(command, str) else " ".join(command) + ) print(self.opts["echo_format"].format(command=command_string)) def _setup(self, command: Union[str, List[str]], kwargs: Any) -> None: @@ -430,7 +434,9 @@ def _setup(self, command: Union[str, List[str]], kwargs: Any) -> None: encoding=self.encoding, ) - def _run_body(self, command: Union[str, List[str]], **kwargs: Any) -> Optional["Result"]: + def _run_body( + self, command: Union[str, List[str]], **kwargs: Any + ) -> Optional["Result"]: # Prepare all the bits n bobs. self._setup(command, kwargs) # If dry-run, stop here. @@ -1045,7 +1051,9 @@ def process_is_finished(self) -> bool: """ raise NotImplementedError - def start(self, command: Union[str, List[str]], shell: str, env: Dict[str, Any]) -> None: + def start( + self, command: Union[str, List[str]], shell: str, env: Dict[str, Any] + ) -> None: """ Initiate execution of ``command`` (via ``shell``, with ``env``). @@ -1307,7 +1315,9 @@ def close_proc_stdin(self) -> None: "Unable to close missing subprocess or stdin!" ) - def start(self, command: Union[str, List[str]], shell: str, env: Dict[str, Any]) -> None: + def start( + self, command: Union[str, List[str]], shell: str, env: Dict[str, Any] + ) -> None: if self.using_pty: if pty is None: # Encountered ImportError err = "You indicated pty=True, but your platform doesn't support the 'pty' module!" # noqa @@ -1334,7 +1344,9 @@ def start(self, command: Union[str, List[str]], shell: str, env: Dict[str, Any]) # for now. # NOTE: stdlib subprocess (actually its posix flavor, which is # written in C) uses either execve or execv, depending. - command_parts = [command] if isinstance(command, str) else command + command_parts = ( + [command] if isinstance(command, str) else command + ) os.execve(shell, [shell, "-c", *command_parts], env) else: self.process = Popen(