Backport: Rework exceptions

1. Allow substitute `CalledProcessError`
2. Modify exceptions class hierarchy

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

Author: penguinolog

doc/source/exceptions.rst

@@ -100,16 +100,14 @@ API: exceptions
         ``typing.Text``
         stdout string or brief string
 
-.. py:exception:: ParallelCallExceptions(ExecCalledProcessError)
+.. py:exception:: ParallelCallProcessError(ExecCalledProcessError)
 
-    Exception raised during parallel call as result of exceptions.
+    Exception during parallel execution.
 
-    .. py:method:: __init__(command, exceptions, errors, results, expected=None, )
+    .. py:method:: __init__(command, errors, results, expected=None, )
 
         :param command: command
         :type command: ``str``
-        :param exceptions: Exception on connections
-        :type exceptions: ``typing.Dict[typing.Tuple[str, int], Exception]``
         :param errors: results with errors
         :type errors: typing.Dict[typing.Tuple[str, int], ExecResult]
         :param results: all results
@@ -124,11 +122,6 @@ API: exceptions
         ``str``
         command
 
-    .. py:attribute:: exceptions
-
-        ``typing.Dict[typing.Tuple[str, int], Exception]``
-        Exception on connections
-
     .. py:attribute:: errors
 
         results with errors
@@ -147,14 +140,16 @@ API: exceptions
 
         :rtype: typing.List[typing.Union[int, ExitCodes]]
 
-.. py:exception:: ParallelCallProcessError(ExecCalledProcessError)
+.. py:exception:: ParallelCallExceptions(ParallelCallProcessError)
 
-    Exception during parallel execution.
+    Exception raised during parallel call as result of exceptions.
 
-    .. py:method:: __init__(command, errors, results, expected=None, )
+    .. py:method:: __init__(command, exceptions, errors, results, expected=None, )
 
         :param command: command
         :type command: ``str``
+        :param exceptions: Exception on connections
+        :type exceptions: ``typing.Dict[typing.Tuple[str, int], Exception]``
         :param errors: results with errors
         :type errors: typing.Dict[typing.Tuple[str, int], ExecResult]
         :param results: all results
@@ -169,6 +164,11 @@ API: exceptions
         ``str``
         command
 
+    .. py:attribute:: exceptions
+
+        ``typing.Dict[typing.Tuple[str, int], Exception]``
+        Exception on connections
+
     .. py:attribute:: errors
 
         results with errors

exec_helpers/__init__.py

@@ -58,6 +58,7 @@
 except pkg_resources.DistributionNotFound:
     # package is not installed, try to get from SCM
     try:
+        # noinspection PyPackageRequirements,PyUnresolvedReferences
         import setuptools_scm  # type: ignore
 
         __version__ = setuptools_scm.get_version()

exec_helpers/_ssh_client_base.py

@@ -637,6 +637,7 @@ def _exec_command(  # type: ignore
 
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         """
+
         def poll_streams():  # type: () -> None
             """Poll FIFO buffers if data available."""
             if async_result.stdout and async_result.interface.recv_ready():
@@ -811,6 +812,7 @@ def execute_together(
         .. versionchanged:: 1.2.0 default timeout 1 hour
         .. versionchanged:: 1.2.0 log_mask_re regex rule for masking cmd
         """
+
         @threaded.threadpooled
         def get_result(remote):  # type: (SSHClientBase) -> exec_result.ExecResult
             """Get result from remote call."""
@@ -858,7 +860,9 @@ def get_result(remote):  # type: (SSHClientBase) -> exec_result.ExecResult
         if raised_exceptions:  # always raise
             raise exceptions.ParallelCallExceptions(command, raised_exceptions, errors, results, expected=expected)
         if errors and raise_on_err:
-            raise exceptions.ParallelCallProcessError(command, errors, results, expected=expected)
+            raise kwargs.get("exception_class", exceptions.ParallelCallProcessError)(
+                command, errors, results, expected=expected
+            )
         return results
 
     def open(self, path, mode="r"):  # type: (str, str) -> paramiko.SFTPFile

exec_helpers/_subprocess_helpers.py

@@ -40,6 +40,7 @@ def kill_proc_tree(pid, including_parent=True):  # type:(int, bool) -> None  # p
     :param including_parent: kill also parent process
     :type including_parent: bool
     """
+
     def safe_stop(proc, kill=False):  # type: (psutil.Process, bool) -> None
         """Do not crash on already stopped process."""
         try:

exec_helpers/api.py

@@ -107,6 +107,7 @@ def _mask_command(self, cmd, log_mask_re=None):  # type: (str, typing.Optional[s
 
         .. versionadded:: 1.2.0
         """
+
         def mask(text, rules):  # type: (str, str) -> str
             """Mask part of text using rules."""
             indexes = [0]  # Start of the line
@@ -125,7 +126,7 @@ def mask(text, rules):  # type: (str, str) -> str
                 end = indexes[idx + 1]
                 masked += text[start:end] + "<*masked*>"
 
-            masked += text[indexes[-2]: indexes[-1]]  # final part
+            masked += text[indexes[-2] : indexes[-1]]  # final part
             return masked
 
         cmd = cmd.rstrip()
@@ -292,7 +293,7 @@ def check_call(
             )
             self.logger.error(msg=message)
             if raise_on_err:
-                raise exceptions.CalledProcessError(result=ret, expected=expected_codes)
+                raise kwargs.get("exception_class", exceptions.CalledProcessError)(result=ret, expected=expected_codes)
         return ret
 
     def check_stderr(
@@ -335,7 +336,9 @@ def check_stderr(
             )
             self.logger.error(msg=message)
             if raise_on_err:
-                raise exceptions.CalledProcessError(result=ret, expected=kwargs.get("expected"))
+                raise kwargs.get("exception_class", exceptions.CalledProcessError)(
+                    result=ret, expected=kwargs.get("expected")
+                )
         return ret
 
     @staticmethod

exec_helpers/exceptions.py

@@ -71,10 +71,10 @@ def __init__(self, result, timeout):  # type: (exec_result.ExecResult, typing.Un
         :param timeout: timeout for command
         :type timeout: typing.Union[int, float]
         """
-        self.result = result
-        self.timeout = timeout
         message = _log_templates.CMD_WAIT_ERROR.format(result=result, timeout=timeout)
         super(ExecHelperTimeoutError, self).__init__(message)
+        self.result = result
+        self.timeout = timeout
 
     @property
     def cmd(self):  # type: () -> str
@@ -100,14 +100,14 @@ class CalledProcessError(ExecCalledProcessError):
     def __init__(
         self,
         result,  # type: exec_result.ExecResult
-        expected=None,  # type: typing.Optional[typing.List[typing.Union[int, proc_enums.ExitCodes]]]
+        expected=None,  # type: typing.Optional[typing.Iterable[typing.Union[int, proc_enums.ExitCodes]]]
     ):  # type: (...) -> None
         """Exception for error on process calls.
 
         :param result: execution result
         :type result: exec_result.ExecResult
         :param expected: expected return codes
-        :type expected: typing.Optional[typing.List[typing.Union[int, proc_enums.ExitCodes]]]
+        :type expected: typing.Optional[typing.Iterable[typing.Union[int, proc_enums.ExitCodes]]]
 
         .. versionchanged:: 1.1.1 - provide full result
         """
@@ -144,91 +144,97 @@ def stderr(self):  # type: () -> typing.Text
         return self.result.stderr_str
 
 
-class ParallelCallExceptions(ExecCalledProcessError):
-    """Exception raised during parallel call as result of exceptions."""
+class ParallelCallProcessError(ExecCalledProcessError):
+    """Exception during parallel execution."""
 
-    __slots__ = ("cmd", "exceptions", "errors", "results", "expected")
+    __slots__ = ("cmd", "errors", "results", "expected")
 
     def __init__(
         self,
         command,  # type: str
-        exceptions,  # type: typing.Dict[typing.Tuple[str, int], Exception]
         errors,  # type: typing.Dict[typing.Tuple[str, int], exec_result.ExecResult]
         results,  # type: typing.Dict[typing.Tuple[str, int], exec_result.ExecResult]
         expected=None,  # type: typing.Optional[typing.List[typing.Union[int, proc_enums.ExitCodes]]]
+        **kwargs  # type: typing.Any
     ):  # type: (...) -> None
         """Exception raised during parallel call as result of exceptions.
 
         :param command: command
         :type command: str
-        :param exceptions: Exceptions on connections
-        :type exceptions: typing.Dict[typing.Tuple[str, int], Exception]
         :param errors: results with errors
         :type errors: typing.Dict[typing.Tuple[str, int], ExecResult]
         :param results: all results
         :type results: typing.Dict[typing.Tuple[str, int], ExecResult]
         :param expected: expected return codes
         :type expected: typing.Optional[typing.List[typing.Union[int, proc_enums.ExitCodes]]]
+        :keyword _message: error message override
         """
         expected = expected or [proc_enums.ExitCodes.EX_OK]
-        self.expected = proc_enums.exit_codes_to_enums(expected)
-        self.cmd = command
-        self.exceptions = exceptions
-        self.errors = errors
-        self.results = results
-        message = (
-            "Command {self.cmd!r} "
-            "during execution raised exceptions: \n"
-            "\t{exceptions}".format(
-                self=self,
-                exceptions="\n\t".join(
-                    "{host}:{port} - {exc} ".format(host=host, port=port, exc=exc)
-                    for (host, port), exc in exceptions.items()
+        prep_expected = proc_enums.exit_codes_to_enums(expected)
+        message = kwargs.get("_message", None) or (
+            "Command {cmd!r} "
+            "returned unexpected exit codes on several hosts\n"
+            "Expected: {expected}\n"
+            "Got:\n"
+            "\t{errors}".format(
+                cmd=command,
+                expected=prep_expected,
+                errors="\n\t".join(
+                    "{host}:{port} - {code} ".format(host=host, port=port, code=result.exit_code)
+                    for (host, port), result in errors.items()
                 ),
             )
         )
-        super(ParallelCallExceptions, self).__init__(message)
+        super(ParallelCallProcessError, self).__init__(message)
+        self.cmd = command
+        self.errors = errors
+        self.results = results
+        self.expected = prep_expected
 
 
-class ParallelCallProcessError(ExecCalledProcessError):
-    """Exception during parallel execution."""
+class ParallelCallExceptions(ParallelCallProcessError):
+    """Exception raised during parallel call as result of exceptions."""
 
-    __slots__ = ("cmd", "errors", "results", "expected")
+    __slots__ = ("cmd", "exceptions")
 
     def __init__(
         self,
         command,  # type: str
+        exceptions,  # type: typing.Dict[typing.Tuple[str, int], Exception]
         errors,  # type: typing.Dict[typing.Tuple[str, int], exec_result.ExecResult]
         results,  # type: typing.Dict[typing.Tuple[str, int], exec_result.ExecResult]
         expected=None,  # type: typing.Optional[typing.List[typing.Union[int, proc_enums.ExitCodes]]]
+        **kwargs  # type: typing.Any
     ):  # type: (...) -> None
         """Exception during parallel execution.
 
         :param command: command
         :type command: str
+        :param exceptions: Exceptions on connections
+        :type exceptions: typing.Dict[typing.Tuple[str, int], Exception]
         :param errors: results with errors
         :type errors: typing.Dict[typing.Tuple[str, int], ExecResult]
         :param results: all results
         :type results: typing.Dict[typing.Tuple[str, int], ExecResult]
         :param expected: expected return codes
         :type expected: typing.Optional[typing.List[typing.Union[int, proc_enums.ExitCodes]]]
+        :keyword _message: error message override
         """
         expected = expected or [proc_enums.ExitCodes.EX_OK]
-        self.expected = proc_enums.exit_codes_to_enums(expected)
-        self.cmd = command
-        self.errors = errors
-        self.results = results
-        message = (
-            "Command {self.cmd!r} "
-            "returned unexpected exit codes on several hosts\n"
-            "Expected: {self.expected}\n"
-            "Got:\n"
-            "\t{errors}".format(
-                self=self,
-                errors="\n\t".join(
-                    "{host}:{port} - {code} ".format(host=host, port=port, code=result.exit_code)
-                    for (host, port), result in errors.items()
+        prep_expected = proc_enums.exit_codes_to_enums(expected)
+        message = kwargs.get("_message", None) or (
+            "Command {cmd!r} "
+            "during execution raised exceptions: \n"
+            "\t{exceptions}".format(
+                cmd=command,
+                exceptions="\n\t".join(
+                    "{host}:{port} - {exc} ".format(host=host, port=port, exc=exc)
+                    for (host, port), exc in exceptions.items()
                 ),
             )
         )
-        super(ParallelCallProcessError, self).__init__(message)
+        super(ParallelCallExceptions, self).__init__(
+            command=command, errors=errors, results=results, expected=prep_expected, _message=message
+        )
+        self.cmd = command
+        self.exceptions = exceptions

exec_helpers/metaclasses.py

@@ -20,8 +20,10 @@
 
 import abc
 import threading
+
 # pylint: disable=unused-import
 import typing  # noqa: F401
+
 # pylint: enable=unused-import
 
 

exec_helpers/subprocess_runner.py

@@ -101,6 +101,7 @@ def _exec_command(  # type: ignore
 
         .. versionadded:: 1.2.0
         """
+
         @threaded.threadpooled
         def poll_stdout():  # type: () -> None
             """Sync stdout poll."""
@@ -149,21 +150,21 @@ def close_streams():  # type: () -> None
             exit_code = async_result.interface.poll()
             if exit_code is not None:  # Nothing to kill
                 self.logger.warning(
-                    "{!s} has been completed just after timeout: please validate timeout.".format(command))
+                    "{!s} has been completed just after timeout: please validate timeout.".format(command)
+                )
                 concurrent.futures.wait([stdout_future, stderr_future], timeout=0.1)
                 result.exit_code = exit_code
                 return result
             raise  # Some other error
         finally:
             stdout_future.cancel()
             stderr_future.cancel()
-            _, not_done = concurrent.futures.wait([stdout_future, stderr_future], timeout=5)
+            _, not_done = concurrent.futures.wait([stdout_future, stderr_future], timeout=1)
             if not_done:
-                exit_code = async_result.interface.poll()
-                if exit_code:
+                if async_result.interface.returncode:
                     self.logger.critical(
                         "Process {!s} was closed with exit code {!s}, but FIFO buffers are still open".format(
-                            command, exit_code
+                            command, async_result.interface.returncode
                         )
                     )
             close_streams()

setup.cfg

@@ -39,15 +39,25 @@ exclude =
     __init__.py,
     docs
 ignore =
-    # line break before binary operator
+    E203,
     W503
+# whitespace before ':'
+# line break before binary operator
 show-pep8 = True
 show-source = True
 count = True
 max-line-length = 120
 
 [pydocstyle]
-ignore = D401, D203, D213
+ignore =
+    D401,
+    D202,
+    D203,
+    D213
+# First line should be in imperative mood; try rephrasing
+# No blank lines allowed after function docstring
+# 1 blank line required before class docstring
+# Multi-line docstring summary should start at the second line
 
 [aliases]
 test=pytest