Commit 2db6987c by Jörg Stucke

added check option

parent 42f97ed0
from subprocess import Popen, PIPE, STDOUT, TimeoutExpired
from signal import SIGKILL
import logging import logging
import pexpect from signal import SIGKILL
from subprocess import Popen, PIPE, STDOUT, TimeoutExpired, CalledProcessError
from time import sleep 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. Execute a shell command and return STDOUT and STDERR in one combined result string.
This function shall not raise any errors This function shall not raise any errors
:param shell_command: command to execute :param shell_command: command to execute
:type shell_command: str :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: timeout: bool
:return: str :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): def execute_shell_command_get_return_code(shell_command, timeout=None):
""" """
Execute a shell command and return a tuple (program output, return code) 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 This function shall not raise any errors
:param shell_command: command to execute :param shell_command: command to execute
:type shell_command: str :type shell_command: str
:param timeout: kill process after timeout seconds :param timeout: kill process after timeout seconds
:type: timeout: int :type: timeout: int, optional
:return: str, int :return: str, int
""" """
output = ""
return_code = 1
pl = Popen(shell_command, shell=True, stdout=PIPE, stderr=STDOUT) pl = Popen(shell_command, shell=True, stdout=PIPE, stderr=STDOUT)
try: try:
output = pl.communicate(timeout=timeout)[0].decode('utf-8', errors='replace') 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): ...@@ -44,7 +50,7 @@ def execute_shell_command_get_return_code(shell_command, timeout=None):
return output, return_code 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) Execute an interactive shell command and return a tuple (program output, return code)
This function shall not raise any errors This function shall not raise any errors
...@@ -54,9 +60,11 @@ def execute_interactive_shell_command(shell_command, timeout=60, inputs={}): ...@@ -54,9 +60,11 @@ def execute_interactive_shell_command(shell_command, timeout=60, inputs={}):
:param timeout: kill process after timeout seconds :param timeout: kill process after timeout seconds
:type timeout: int :type timeout: int
:param inputs: dictionary {'EXPECTED_CONSOLE_OUTPUT': 'DESIRED_INPUT'} :param inputs: dictionary {'EXPECTED_CONSOLE_OUTPUT': 'DESIRED_INPUT'}
:type inputs: dict :type inputs: dict, optional
:return: str, int :return: str, int
""" """
if inputs is None:
inputs = {}
trigger, inputs = _parse_inputs(inputs) trigger, inputs = _parse_inputs(inputs)
output = b'' output = b''
child = pexpect.spawn(shell_command) child = pexpect.spawn(shell_command)
......
from setuptools import setup, find_packages from setuptools import setup, find_packages
VERSION = 0.3 VERSION = 0.4
setup( setup(
name="common_helper_process", name="common_helper_process",
......
import os
import unittest import unittest
from time import time from time import time
import os
from common_helper_files import get_dir_of_file 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 from common_helper_process.fail_safe_subprocess import _parse_inputs
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment