Implement asyncio API for subprocess (#95)

[x] API mimic sync version
[x] partial migration to pytest for parametrized tests
[x] Fix no stdin in exec result

Author: penguinolog

.gitignore

@@ -6,7 +6,7 @@
 .pytest_cache/*
 
 ### Generated code
-/exec_helpers/*.c
+/exec_helpers/**/*.c
 
 ### TortoiseGit template
 # Project-level settings

.travis.yml

@@ -3,14 +3,11 @@ language: python
 os: linux
 install:
 - &upgrade_python_toolset pip install --upgrade pip setuptools wheel
-- &install_test_deps pip install --upgrade pytest pytest-sugar
+- &install_test_deps pip install --upgrade asynctest mock pytest pytest-mock pytest-asyncio pytest-sugar
 - &install_deps pip install -r CI_REQUIREMENTS.txt
 - pip install --upgrade pytest-cov coveralls
 
 _python:
-- &python34
-  name: "Python 3.4"
-  python: 3.4
 - &python35
   name: "Python 3.5"
   python: 3.5
@@ -110,8 +107,6 @@ jobs:
     script:
     - black --check exec_helpers
 
-  - stage: test
-    <<: *python34
   - stage: test
     <<: *python35
   - stage: test
@@ -121,8 +116,6 @@ jobs:
   - stage: test
     <<: *pypy3
 
-  - <<: *test_cythonized
-    <<: *python34
   - <<: *test_cythonized
     <<: *python35
   - <<: *test_cythonized

README.rst

@@ -44,13 +44,12 @@ Pros:
 
 ::
 
-    Python 3.4
     Python 3.5
     Python 3.6
     Python 3.7
     PyPy3 3.5+
 
-.. note:: For Python 2.7 and PyPy please use versions 1.x.x
+.. note:: For Python 2.7 and PyPy please use versions 1.x.x. For python 3.4 use versions 2.x.x
 
 This package includes:
 
@@ -64,6 +63,9 @@ This package includes:
 * `Subprocess` - `subprocess.Popen` wrapper with timeouts, polling and almost the same API, as `SSHClient`
   (except specific flags, like `cwd` for subprocess and `get_tty` for ssh).
 
+* `async_api.Subprocess` - the same, as `Subprocess` helper, but works with asyncio.
+  .. note:: for Windows `ProactorEventLoop` or another non-standard event loop should be used!
+
 * `ExecResult` - class for execution results storage.
   Contains exit code, stdout, stderr and getters for decoding as JSON, YAML, string, bytearray and brief strings (up to 7 lines).
 
@@ -315,6 +317,23 @@ Kwargs set properties:
 
 .. note:: `shell=true` is always set.
 
+async_api.Subprocess specific
+-----------------------------
+
+All standard methods are coroutines. Async context manager also available.
+
+Example:
+
+.. code-block:: python
+
+    async with helper:
+      result = await helper.execute(
+          command,  # type: str
+          verbose=False,  # type: bool
+          timeout=1 * 60 * 60,  # type: typing.Union[int, float, None]
+          **kwargs
+      )
+
 Testing
 =======
 The main test mechanism for the package `exec-helpers` is using `tox`.

build_requirements.txt

@@ -1,4 +1,4 @@
 Cython; platform_python_implementation == "CPython"
-wheel
+wheel==0.31.1
 -r CI_REQUIREMENTS.txt
 -r requirements.txt

doc/source/ExecResult.rst

@@ -6,7 +6,7 @@ API: ExecResult
 .. py:module:: exec_helpers
 .. py:currentmodule:: exec_helpers
 
-.. py:class:: ExecResult(object)
+.. py:class:: ExecResult()
 
     Command execution result.
 
@@ -27,6 +27,7 @@ API: ExecResult
 
         ``threading.RLock``
         Lock object for thread-safe operation.
+
         .. versionadded:: 2.2.0
 
     .. py:attribute:: stderr_lock

doc/source/SSHClient.rst

@@ -325,7 +325,7 @@ API: SSHClient and SSHAuth.
         :rtype: ``bool``
 
 
-.. py:class:: SSHAuth(object)
+.. py:class:: SSHAuth()
 
     SSH credentials object.
 

exec_helpers/__init__.py

@@ -25,12 +25,13 @@
     ExecHelperTimeoutError,
 )
 
-from .exec_result import ExecResult
 from .api import ExecHelper
+from .exec_result import ExecResult
 from .ssh_auth import SSHAuth
-from ._ssh_client_base import SshExecuteAsyncResult
 from .ssh_client import SSHClient
+from ._ssh_client_base import SshExecuteAsyncResult
 from .subprocess_runner import Subprocess, SubprocessExecuteAsyncResult  # nosec  # Expected
+from . import async_api
 
 __all__ = (
     "ExecHelperError",
@@ -47,9 +48,10 @@
     "SubprocessExecuteAsyncResult",
     "ExitCodes",
     "ExecResult",
+    "async_api",
 )
 
-__version__ = "2.2.0"
+__version__ = "3.0.0"
 __author__ = "Alexey Stepanov"
 __author_email__ = "penguinolog@gmail.com"
 __maintainers__ = {

exec_helpers/_ssh_client_base.py

@@ -614,12 +614,10 @@ def execute_async(
 
         return SshExecuteAsyncResult(chan, _stdin, stderr, stdout)
 
-    def _exec_command(
+    def _exec_command(  # type: ignore
         self,
         command: str,
-        interface: paramiko.channel.Channel,
-        stdout: typing.Optional[paramiko.ChannelFile],
-        stderr: typing.Optional[paramiko.ChannelFile],
+        async_result: SshExecuteAsyncResult,
         timeout: typing.Union[int, float, None],
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
@@ -629,12 +627,8 @@ def _exec_command(
 
         :param command: executed command (for logs)
         :type command: str
-        :param interface: interface to control execution
-        :type interface: paramiko.channel.Channel
-        :param stdout: source for STDOUT read
-        :type stdout: typing.Optional[paramiko.ChannelFile]
-        :param stderr: source for STDERR read
-        :type stderr: typing.Optional[paramiko.ChannelFile]
+        :param async_result: execute_async result
+        :type async_result: SshExecuteAsyncResult
         :param timeout: timeout before stop execution with TimeoutError
         :type timeout: typing.Union[int, float, None]
         :param verbose: produce log.info records for STDOUT/STDERR
@@ -653,10 +647,10 @@ def _exec_command(
 
         def poll_streams() -> None:
             """Poll FIFO buffers if data available."""
-            if stdout and interface.recv_ready():
-                result.read_stdout(src=stdout, log=self.logger, verbose=verbose)
-            if stderr and interface.recv_stderr_ready():
-                result.read_stderr(src=stderr, log=self.logger, verbose=verbose)
+            if async_result.stdout and async_result.interface.recv_ready():
+                result.read_stdout(src=async_result.stdout, log=self.logger, verbose=verbose)
+            if async_result.stderr and async_result.interface.recv_stderr_ready():
+                result.read_stderr(src=async_result.stderr, log=self.logger, verbose=verbose)
 
         @threaded.threadpooled
         def poll_pipes(stop: threading.Event) -> None:
@@ -666,21 +660,21 @@ def poll_pipes(stop: threading.Event) -> None:
             """
             while not stop.is_set():
                 time.sleep(0.1)
-                if stdout or stderr:
+                if async_result.stdout or async_result.stderr:
                     poll_streams()
 
-                if interface.status_event.is_set():
-                    result.read_stdout(src=stdout, log=self.logger, verbose=verbose)
-                    result.read_stderr(src=stderr, log=self.logger, verbose=verbose)
-                    result.exit_code = interface.exit_status
+                if async_result.interface.status_event.is_set():
+                    result.read_stdout(src=async_result.stdout, log=self.logger, verbose=verbose)
+                    result.read_stderr(src=async_result.stderr, log=self.logger, verbose=verbose)
+                    result.exit_code = async_result.interface.exit_status
 
                     stop.set()
 
         # channel.status_event.wait(timeout)
         cmd_for_log = self._mask_command(cmd=command, log_mask_re=log_mask_re)
 
         # Store command with hidden data
-        result = exec_result.ExecResult(cmd=cmd_for_log)
+        result = exec_result.ExecResult(cmd=cmd_for_log, stdin=kwargs.get("stdin"))
 
         stop_event = threading.Event()
 
@@ -693,11 +687,11 @@ def poll_pipes(stop: threading.Event) -> None:
 
         # Process closed?
         if stop_event.is_set():
-            interface.close()
+            async_result.interface.close()
             return result
 
         stop_event.set()
-        interface.close()
+        async_result.interface.close()
         future.cancel()
 
         wait_err_msg = _log_templates.CMD_WAIT_ERROR.format(result=result, timeout=timeout)
@@ -774,9 +768,15 @@ def execute_through_host(
 
         channel.exec_command(command)  # nosec  # Sanitize on caller side
 
+        async_result = SshExecuteAsyncResult(interface=channel, stdin=None, stdout=stdout, stderr=stderr)
+
         # noinspection PyDictCreation
         result = self._exec_command(
-            command, channel, stdout, stderr, timeout, verbose=verbose, log_mask_re=kwargs.get("log_mask_re", None)
+            command,
+            async_result=async_result,
+            timeout=timeout,
+            verbose=verbose,
+            log_mask_re=kwargs.get("log_mask_re", None),
         )
 
         intermediate_channel.close()

exec_helpers/_subprocess_helpers.py

@@ -0,0 +1,58 @@
+#    Copyright 2018 Alexey Stepanov aka penguinolog.
+#
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+"""Python subprocess shared code."""
+
+__all__ = ("kill_proc_tree", "subprocess_kw")
+
+import platform
+
+# pylint: disable=unused-import
+import typing  # noqa: F401
+
+# pylint: enable=unused-import
+
+import psutil  # type: ignore
+
+
+# Adopt from:
+# https://stackoverflow.com/questions/1230669/subprocess-deleting-child-processes-in-windows
+def kill_proc_tree(pid: int, including_parent: bool = True) -> None:  # pragma: no cover
+    """Kill process tree.
+
+    :param pid: PID of parent process to kill
+    :type pid: int
+    :param including_parent: kill also parent process
+    :type including_parent: bool
+    """
+    parent = psutil.Process(pid)
+    children = parent.children(recursive=True)
+    for child in children:  # type: psutil.Process
+        child.kill()
+    _, alive = psutil.wait_procs(children, timeout=5)
+    for proc in alive:  # type: psutil.Process
+        proc.kill()  # 2nd shot
+    if including_parent:
+        parent.kill()
+        parent.wait(5)
+
+
+# Subprocess extra arguments.
+# Flags from:
+# https://stackoverflow.com/questions/13243807/popen-waiting-for-child-process-even-when-the-immediate-child-has-terminated
+subprocess_kw = {}  # type: typing.Dict[str, typing.Any]
+if "Windows" == platform.system():  # pragma: no cover
+    subprocess_kw["creationflags"] = 0x00000200
+else:  # pragma: no cover
+    subprocess_kw["start_new_session"] = True

exec_helpers/api.py

@@ -184,9 +184,7 @@ def execute_async(
     def _exec_command(
         self,
         command: str,
-        interface: typing.Any,
-        stdout: typing.Any,
-        stderr: typing.Any,
+        async_result: ExecuteAsyncResult,
         timeout: typing.Union[int, float, None],
         verbose: bool = False,
         log_mask_re: typing.Optional[str] = None,
@@ -196,12 +194,8 @@ def _exec_command(
 
         :param command: Command for execution
         :type command: str
-        :param interface: Control interface
-        :type interface: typing.Any
-        :param stdout: STDOUT pipe or file-like object
-        :type stdout: typing.Any
-        :param stderr: STDERR pipe or file-like object
-        :type stderr: typing.Any
+        :param async_result: execute_async result
+        :type async_result: SubprocessExecuteAsyncResult
         :param timeout: Timeout for command execution
         :type timeout: typing.Union[int, float, None]
         :param verbose: produce verbose log record on command call
@@ -246,13 +240,7 @@ def execute(
         async_result = self.execute_async(command, verbose=verbose, **kwargs)  # type: ExecuteAsyncResult
 
         result = self._exec_command(
-            command=command,
-            interface=async_result.interface,
-            stdout=async_result.stdout,
-            stderr=async_result.stderr,
-            timeout=timeout,
-            verbose=verbose,
-            **kwargs
+            command=command, async_result=async_result, timeout=timeout, verbose=verbose, **kwargs
         )
         message = "Command {result.cmd!r} exit code: {result.exit_code!s}".format(result=result)
         self.logger.log(level=logging.INFO if verbose else logging.DEBUG, msg=message)  # type: ignore
@@ -300,7 +288,7 @@ def check_call(
                     append=error_info + "\n" if error_info else "", result=ret, expected=expected
                 )
             )
-            self.logger.error(message)
+            self.logger.error(msg=message)
             if raise_on_err:
                 raise exceptions.CalledProcessError(result=ret, expected=expected)
         return ret
@@ -340,10 +328,10 @@ def check_stderr(
         )
         if ret.stderr:
             message = (
-                "{append}Command {result.cmd!r} STDERR while not expected\n"
+                "{append}Command {result.cmd!r} output contains STDERR while not expected\n"
                 "\texit code: {result.exit_code!s}".format(append=error_info + "\n" if error_info else "", result=ret)
             )
-            self.logger.error(message)
+            self.logger.error(msg=message)
             if raise_on_err:
                 raise exceptions.CalledProcessError(result=ret, expected=kwargs.get("expected"))
         return ret