"""Utility functions for command execution"""
import logging
import subprocess
from perfmon.exceptions import CommandExecutionFailed
_log = logging.getLogger(__name__)
# pylint: disable=E0401,W0201,C0301
[docs]def execute_cmd(cmd_str, handle_exception=True):
"""Accept command string and returns output.
Args:
cmd_str (str): Command string to be executed
handle_exception (bool): Handle exception manually. If set to false, raises an exception
to the caller function
Returns:
str: Output of the command. If command execution fails, returns 'not_available'
Raises:
subprocess.CalledProcessError: An error occurred in execution of command iff
handle_exception is set to False
"""
# _log.debug("Executing command: %s", cmd_str)
try:
# Execute command
cmd_out = subprocess.run(
cmd_str, shell=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, check=True
)
# Get stdout and stderr. We are piping stderr to stdout as well
cmd_out = cmd_out.stdout.decode('utf-8').rstrip()
except subprocess.CalledProcessError as err:
# If handle_exception is True, return 'not_available'
if handle_exception:
cmd_out = 'not_available'
else:
# If handle_exception is False, raise an exception
_log.warning('Execution of command %s failed', cmd_str)
raise CommandExecutionFailed(
'Execution of command \'{}\' command'.format(cmd_str)
) from err
return cmd_out
[docs]def execute_cmd_pipe(cmd_str):
"""Accept command string and execute it using piping and returns process object.
Args:
cmd_str (str): Command string to be executed
Returns:
object: Process object
"""
proc = subprocess.Popen(
cmd_str,
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
bufsize=-1,
universal_newlines=True,
)
return proc