Deprecate execute_async

Low level method, which normally should not be used from outside: we can change backend without affecting other API parts.

Signed-off-by: Aleksei Stepanov <penguinolog@gmail.com>

Author: penguinolog

exec_helpers/_ssh_client_base.py

@@ -400,7 +400,7 @@ def keepalive(self, enforce: bool = True) -> "typing.ContextManager[None]":
         return _KeepAliveContext(ssh=self, enforce=enforce)
 
     # noinspection PyMethodOverriding
-    def execute_async(  # pylint: disable=arguments-differ
+    def _execute_async(  # pylint: disable=arguments-differ
         self,
         command: str,
         stdin: typing.Union[bytes, str, bytearray, None] = None,
@@ -743,7 +743,7 @@ def get_result(remote: "SSHClientBase") -> exec_result.ExecResult:
             :param remote: SSH connection instance
             :returns: execution result
             """
-            async_result: SshExecuteAsyncResult = remote.execute_async(
+            async_result: SshExecuteAsyncResult = remote._execute_async(  # pylint: disable=protected-access
                 command, stdin=stdin, log_mask_re=log_mask_re, **kwargs
             )
 

exec_helpers/api.py

@@ -28,6 +28,7 @@
 import re
 import threading
 import typing
+import warnings
 
 # Exec-Helpers Implementation
 from exec_helpers import constants
@@ -229,8 +230,53 @@ def _prepare_command(self, cmd: str, chroot_path: typing.Optional[str] = None) -
             return f"chroot {chroot_path if chroot_path else self._chroot_path} {cmd}"
         return cmd
 
+    def execute_async(  # pylint: disable=missing-param-doc,differing-param-doc,differing-type-doc
+        self, *args: typing.Any, **kwargs: typing.Any
+    ) -> ExecuteAsyncResult:
+        """Execute command in async mode and return remote interface with IO objects.
+
+        :param command: Command for execution
+        :type command: str
+        :param stdin: pass STDIN text to the process
+        :type stdin: typing.Union[bytes, str, bytearray, None]
+        :param open_stdout: open STDOUT stream for read
+        :type open_stdout: bool
+        :param open_stderr: open STDERR stream for read
+        :type open_stderr: bool
+        :param verbose: produce verbose log record on command call
+        :type verbose: bool
+        :param log_mask_re: regex lookup rule to mask command for logger.
+                            all MATCHED groups will be replaced by '<*masked*>'
+        :type log_mask_re: typing.Optional[str]
+        :param chroot_path: chroot path override
+        :type chroot_path: typing.Optional[str]
+        :param kwargs: additional parameters for call.
+        :type kwargs: typing.Any
+        :return: NamedTuple with control interface and file-like objects for STDIN/STDERR/STDOUT
+        :rtype: typing.NamedTuple(
+                    'ExecuteAsyncResult',
+                    [
+                        ('interface', typing.Any),
+                        ('stdin', typing.Optional[typing.Any]),
+                        ('stderr', typing.Optional[typing.Any]),
+                        ('stdout', typing.Optional[typing.Any]),
+                        ("started", datetime.datetime),
+                    ]
+                )
+
+        .. versionchanged:: 1.2.0 open_stdout and open_stderr flags
+        .. versionchanged:: 1.2.0 stdin data
+        .. versionchanged:: 2.1.0 Use typed NamedTuple as result
+        .. versionchanged:: 4.1.0 support chroot
+        """
+        warnings.warn(
+            "execute_async public usage is deprecated and will be disallowed at the next major release.",
+            DeprecationWarning,
+        )
+        return self._execute_async(*args, **kwargs)
+
     @abc.abstractmethod
-    def execute_async(
+    def _execute_async(
         self,
         command: str,
         stdin: typing.Union[bytes, str, bytearray, None] = None,
@@ -347,7 +393,7 @@ def execute(
         .. versionchanged:: 1.2.0 default timeout 1 hour
         .. versionchanged:: 2.1.0 Allow parallel calls
         """
-        async_result: ExecuteAsyncResult = self.execute_async(
+        async_result: ExecuteAsyncResult = self._execute_async(
             command, verbose=verbose, log_mask_re=log_mask_re, stdin=stdin, **kwargs
         )
 

exec_helpers/async_api/api.py

@@ -134,7 +134,7 @@ async def _exec_command(  # type: ignore
         """
 
     @abc.abstractmethod
-    async def execute_async(  # type: ignore
+    async def _execute_async(  # type: ignore
         self,
         command: str,
         stdin: typing.Union[str, bytes, bytearray, None] = None,
@@ -208,7 +208,7 @@ async def execute(  # type: ignore
         :rtype: ExecResult
         :raises ExecHelperTimeoutError: Timeout exceeded
         """
-        async_result: api.ExecuteAsyncResult = await self.execute_async(
+        async_result: api.ExecuteAsyncResult = await self._execute_async(
             command, verbose=verbose, log_mask_re=log_mask_re, stdin=stdin, **kwargs
         )
 

exec_helpers/async_api/subprocess_runner.py

@@ -163,7 +163,7 @@ async def poll_stderr() -> None:
 
     # pylint: disable=arguments-differ
     # noinspection PyMethodOverriding
-    async def execute_async(  # type: ignore
+    async def _execute_async(  # type: ignore
         self,
         command: str,
         stdin: typing.Union[str, bytes, bytearray, None] = None,

exec_helpers/subprocess_runner.py

@@ -169,12 +169,11 @@ def close_streams() -> None:
             stdout_future.cancel()
             stderr_future.cancel()
             _, not_done = concurrent.futures.wait([stdout_future, stderr_future], timeout=1)
-            if not_done:
-                if async_result.interface.returncode:
-                    self.logger.critical(
-                        f"Process {command!s} was closed with exit code {async_result.interface.returncode!s}, "
-                        f"but FIFO buffers are still open"
-                    )
+            if not_done and async_result.interface.returncode:
+                self.logger.critical(
+                    f"Process {command!s} was closed with exit code {async_result.interface.returncode!s}, "
+                    f"but FIFO buffers are still open"
+                )
             result.set_timestamp()
             close_streams()
 
@@ -183,7 +182,7 @@ def close_streams() -> None:
         raise exceptions.ExecHelperTimeoutError(result=result, timeout=timeout)  # type: ignore
 
     # noinspection PyMethodOverriding
-    def execute_async(  # pylint: disable=arguments-differ
+    def _execute_async(  # pylint: disable=arguments-differ
         self,
         command: str,
         stdin: typing.Union[str, bytes, bytearray, None] = None,

test/async_api/test_subprocess.py

@@ -202,7 +202,7 @@ def logger(mocker):
 async def test_001_execute_async(create_subprocess_shell, logger, run_parameters) -> None:
     """Test low level API."""
     runner = exec_helpers.async_api.Subprocess()
-    res = await runner.execute_async(
+    res = await runner._execute_async(
         command,
         stdin=run_parameters["stdin"],
         open_stdout=run_parameters["open_stdout"],

test/test_ssh_client_execute.py

@@ -235,7 +235,7 @@ def get_patched_execute_async_retval(
         )
 
     return mocker.patch(
-        "exec_helpers.ssh_client.SSHClient.execute_async",
+        "exec_helpers.ssh_client.SSHClient._execute_async",
         side_effect=[
             get_patched_execute_async_retval(**run_parameters),
             get_patched_execute_async_retval(**run_parameters),
@@ -261,7 +261,7 @@ def test_001_execute_async(ssh, paramiko_ssh_client, ssh_transport_channel, chan
     if "height" in run_parameters:
         kwargs["height"] = run_parameters["height"]
 
-    res = ssh.execute_async(
+    res = ssh._execute_async(
         command, stdin=run_parameters["stdin"], open_stdout=open_stdout, open_stderr=open_stderr, **kwargs
     )
     assert isinstance(res, exec_helpers.SshExecuteAsyncResult)

test/test_ssh_client_execute_async_special.py

@@ -116,7 +116,7 @@ def exec_result():
 def test_001_execute_async_sudo(ssh, ssh_transport_channel):
     ssh.sudo_mode = True
 
-    ssh.execute_async(command)
+    ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
@@ -129,11 +129,11 @@ def test_002_execute_async_with_sudo_enforce(ssh, ssh_transport_channel):
     assert ssh.sudo_mode is False
 
     with ssh.sudo(enforce=True):
-        ssh.execute_async(command)
+        ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'sudo -S bash -c \'eval "$(base64 -d <(echo "{encoded_cmd}"))"\''),
+            mock.call.exec_command(f'sudo -S bash -c \"eval {shlex.quote(cmd_execute)}\"'),
         )
     )
 
@@ -142,15 +142,15 @@ def test_003_execute_async_with_no_sudo_enforce(ssh, ssh_transport_channel):
     ssh.sudo_mode = True
 
     with ssh.sudo(enforce=False):
-        ssh.execute_async(command)
+        ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls((mock.call.makefile_stderr("rb"), mock.call.exec_command(f"{command}\n")))
 
 
 def test_004_execute_async_with_sudo_none_enforce(ssh, ssh_transport_channel):
     ssh.sudo_mode = False
 
     with ssh.sudo():
-        ssh.execute_async(command)
+        ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls((mock.call.makefile_stderr("rb"), mock.call.exec_command(f"{command}\n")))
 
 
@@ -159,11 +159,11 @@ def test_005_execute_async_sudo_password(ssh, ssh_transport_channel, mocker):
 
     ssh.sudo_mode = True
 
-    res = ssh.execute_async(command)
+    res = ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
-            mock.call.exec_command(f'sudo -S bash -c \'eval "$(base64 -d <(echo "{encoded_cmd}"))"\''),
+            mock.call.exec_command(f'sudo -S bash -c \"eval {shlex.quote(cmd_execute)}\"'),
         )
     )
 
@@ -221,15 +221,15 @@ def test_010_check_stdin_closed(paramiko_ssh_client, chan_makefile, auto_add_pol
     stdin_val = "this is a line"
 
     ssh = exec_helpers.SSHClient(host=host, port=port, auth=exec_helpers.SSHAuth(username=username, password=password))
-    ssh.execute_async(command=print_stdin, stdin=stdin_val)
+    ssh._execute_async(command=print_stdin, stdin=stdin_val)
 
     log = get_logger(ssh.__class__.__name__).getChild(f"{host}:{port}")
     log.warning.assert_called_once_with("STDIN Send failed: closed channel")
 
 
 def test_011_execute_async_chroot_cmd(ssh, ssh_transport_channel):
     """Command-only chroot path."""
-    ssh.execute_async(command, chroot_path='/')
+    ssh._execute_async(command, chroot_path='/')
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
@@ -241,7 +241,7 @@ def test_011_execute_async_chroot_cmd(ssh, ssh_transport_channel):
 def test_012_execute_async_chroot_context(ssh, ssh_transport_channel):
     """Context-managed chroot path."""
     with ssh.chroot('/'):
-        ssh.execute_async(command)
+        ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),
@@ -255,7 +255,7 @@ def test_013_execute_async_no_chroot_context(ssh, ssh_transport_channel):
     ssh._chroot_path = "/"
 
     with ssh.chroot(None):
-        ssh.execute_async(command)
+        ssh._execute_async(command)
     ssh_transport_channel.assert_has_calls(
         (
             mock.call.makefile_stderr("rb"),

test/test_ssh_client_execute_special.py

@@ -183,7 +183,7 @@ def test_005_execute_timeout_fail(ssh, ssh_transport_channel, exec_result) -> No
 
 
 def test_006_execute_together_exceptions(ssh, ssh2, mocker) -> None:
-    mocker.patch("exec_helpers.ssh_client.SSHClient.execute_async", side_effect=RuntimeError)
+    mocker.patch("exec_helpers.ssh_client.SSHClient._execute_async", side_effect=RuntimeError)
     remotes = [ssh, ssh2]
 
     with pytest.raises(exec_helpers.ParallelCallExceptions) as e:

test/test_subprocess.py

@@ -177,7 +177,7 @@ def create_mock(
 def test_001_execute_async(popen, subprocess_logger, run_parameters) -> None:
     """Test low level API."""
     runner = exec_helpers.Subprocess()
-    res = runner.execute_async(
+    res = runner._execute_async(
         command,
         stdin=run_parameters["stdin"],
         open_stdout=run_parameters["open_stdout"],