Merge pull request #2 from fkie-cad/add_check_result_option

added check option

.travis.yml

@@ -4,7 +4,9 @@ python:
   - "3.6"
 # command to install dependencies
 install:
-  - "python setup.py -q install"
+  - "pip install git+https://github.com/fkie-cad/common_helper_files.git"
+  - "pip install ."
+  - "pip install -U pytest"
   - "pip install pytest-pep8 pytest-cov codecov"
 # command to run tests
 script: "pytest --cov-config .coveragerc --cov=./"

common_helper_process/fail_safe_subprocess.py

-from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
-from signal import SIGKILL
 import logging
-import pexpect
+from signal import SIGKILL
+from subprocess import Popen, PIPE, STDOUT, TimeoutExpired, CalledProcessError
 from time import sleep
 
+import pexpect
 
-def execute_shell_command(shell_command, timeout=None):
+
+def execute_shell_command(shell_command, timeout=None, check=False):
     """
     Execute a shell command and return STDOUT and STDERR in one combined result string.
     This function shall not raise any errors
 
     :param shell_command: command to execute
     :type shell_command: str
+    :param timeout: kill process after timeout seconds
+    :type: timeout: int, optional
+    :param check: raise CalledProcessError if the return code is != 0
+    :type: check: bool
     :return: str
     """
-    return execute_shell_command_get_return_code(shell_command, timeout=timeout)[0]
+    output, return_code = execute_shell_command_get_return_code(shell_command, timeout=timeout)
+    if check and return_code != 0:
+        raise CalledProcessError(return_code, shell_command)
+    return output
 
 
 def execute_shell_command_get_return_code(shell_command, timeout=None):
     """
     Execute a shell command and return a tuple (program output, return code)
-    Program ouput includes STDOUT and STDERR.
+    Program output includes STDOUT and STDERR.
     This function shall not raise any errors
 
     :param shell_command: command to execute
     :type shell_command: str
     :param timeout: kill process after timeout seconds
-    :type: timeout: int
+    :type: timeout: int, optional
     :return: str, int
     """
-    output = ""
-    return_code = 1
     pl = Popen(shell_command, shell=True, stdout=PIPE, stderr=STDOUT)
     try:
         output = pl.communicate(timeout=timeout)[0].decode('utf-8', errors='replace')
@@ -44,7 +50,7 @@ def execute_shell_command_get_return_code(shell_command, timeout=None):
     return output, return_code
 
 
-def execute_interactive_shell_command(shell_command, timeout=60, inputs={}):
+def execute_interactive_shell_command(shell_command, timeout=60, inputs=None):
     """
     Execute an interactive shell command and return a tuple (program output, return code)
     This function shall not raise any errors
@@ -54,9 +60,11 @@ def execute_interactive_shell_command(shell_command, timeout=60, inputs={}):
     :param timeout: kill process after timeout seconds
     :type timeout: int
     :param inputs: dictionary {'EXPECTED_CONSOLE_OUTPUT': 'DESIRED_INPUT'}
-    :type inputs: dict
+    :type inputs: dict, optional
     :return: str, int
     """
+    if inputs is None:
+        inputs = {}
     trigger, inputs = _parse_inputs(inputs)
     output = b''
     child = pexpect.spawn(shell_command)

setup.py

 from setuptools import setup, find_packages
 
-VERSION = 0.3
+VERSION = 0.4
 
 setup(
     name="common_helper_process",

test/test_fail_safe_subprocess.py

+import os
 import unittest
+from subprocess import CalledProcessError
 from time import time
-import os
+
 from common_helper_files import get_dir_of_file
 
-from common_helper_process import execute_shell_command, execute_shell_command_get_return_code, execute_interactive_shell_command
+from common_helper_process import (
+    execute_interactive_shell_command, execute_shell_command, execute_shell_command_get_return_code
+)
 from common_helper_process.fail_safe_subprocess import _parse_inputs
 
 
@@ -27,6 +31,12 @@ class TestProcessHelperPublic(unittest.TestCase):
         self.assertEqual(output, 'test 123\n', 'result not correct')
         self.assertEqual(rc, 2, 'return code not correct')
 
+    def test_execute_shell_command_with_check(self):
+        with self.assertRaises(CalledProcessError):
+            execute_shell_command("exit 1", check=True)
+        with self.assertRaises(CalledProcessError):
+            execute_shell_command("exit 255", check=True)
+
     def test_execute_shell_command_time_out(self):
         start_time = time()
         output, rc = execute_shell_command_get_return_code('echo \'test 123\' && for i in {1..10}; do sleep 1; done', timeout=1)