"""Robot Framework test library for running processes.

ROBOT_LIBRARY_SCOPE = 'GLOBAL'

ROBOT_LIBRARY_VERSION = get_version()

TERMINATE_TIMEOUT = 30

KILL_TIMEOUT = 10

def __init__(self):

self._processes = ConnectionCache('No active process.')

self._results = {}

def run_process(self, command, *arguments, **configuration):

"""Runs a process and waits for it to complete.

current = self._processes.current

timeout = configuration.pop('timeout', None)

on_timeout = configuration.pop('on_timeout', 'terminate')

try:

handle = self.start_process(command, *arguments, **configuration)

return self.wait_for_process(handle, timeout, on_timeout)

finally:

self._processes.current = current

def start_process(self, command, *arguments, **configuration):

"""Starts a new process on background.

conf = ProcessConfiguration(**configuration)

command = conf.get_command(command, list(arguments))

self._log_start(command, conf)

process = subprocess.Popen(command, **conf.popen_config)

self._results[process] = ExecutionResult(process, **conf.result_config)

return self._processes.register(process, alias=conf.alias)

def _log_start(self, command, config):

if is_list_like(command):

command = self.join_command_line(command)

logger.info('Starting process:\n%s' % system_decode(command))

logger.debug('Process configuration:\n%s' % config)

def is_process_running(self, handle=None):

"""Checks is the process running or not.

return self._processes[handle].poll() is None

def process_should_be_running(self, handle=None,

error_message='Process is not running.'):

"""Verifies that the process is running.

if not self.is_process_running(handle):

raise AssertionError(error_message)

def process_should_be_stopped(self, handle=None,

error_message='Process is running.'):

"""Verifies that the process is not running.

if self.is_process_running(handle):

raise AssertionError(error_message)

def wait_for_process(self, handle=None, timeout=None, on_timeout='continue'):

"""Waits for the process to complete or to reach the given timeout.

process = self._processes[handle]

logger.info('Waiting for process to complete.')

timeout = self._get_timeout(timeout)

if timeout > 0:

if not self._process_is_stopped(process, timeout):

logger.info('Process did not complete in %s.'

% secs_to_timestr(timeout))

return self._manage_process_timeout(handle, on_timeout.lower())

return self._wait(process)

def _get_timeout(self, timeout):

if (is_string(timeout) and timeout.upper() == 'NONE') or not timeout:

return -1

return timestr_to_secs(timeout)

def _manage_process_timeout(self, handle, on_timeout):

if on_timeout == 'terminate':

return self.terminate_process(handle)

elif on_timeout == 'kill':

return self.terminate_process(handle, kill=True)

else:

logger.info('Leaving process intact.')

return None

def _wait(self, process):

result = self._results[process]

result.rc = process.wait() or 0

result.close_streams()

logger.info('Process completed.')

return result

def terminate_process(self, handle=None, kill=False):

"""Stops the process gracefully or forcefully.

process = self._processes[handle]

if not hasattr(process, 'terminate'):

raise RuntimeError('Terminating processes is not supported '

'by this Python version.')

terminator = self._kill if is_truthy(kill) else self._terminate

try:

terminator(process)

except OSError:

if not self._process_is_stopped(process, self.KILL_TIMEOUT):

raise

logger.debug('Ignored OSError because process was stopped.')

return self._wait(process)

def _kill(self, process):

logger.info('Forcefully killing process.')

if hasattr(os, 'killpg'):

os.killpg(process.pid, signal_module.SIGKILL)

else:

process.kill()

if not self._process_is_stopped(process, self.KILL_TIMEOUT):

raise RuntimeError('Failed to kill process.')

def _terminate(self, process):

logger.info('Gracefully terminating process.')

# Sends signal to the whole process group both on POSIX and on Windows

# if supported by the interpreter.

if hasattr(os, 'killpg'):

os.killpg(process.pid, signal_module.SIGTERM)

elif hasattr(signal_module, 'CTRL_BREAK_EVENT'):

process.send_signal(signal_module.CTRL_BREAK_EVENT)

else:

process.terminate()

if not self._process_is_stopped(process, self.TERMINATE_TIMEOUT):

logger.info('Graceful termination failed.')

self._kill(process)

def terminate_all_processes(self, kill=False):

"""Terminates all still running processes started by this library.

for handle in range(1, len(self._processes) + 1):

if self.is_process_running(handle):

self.terminate_process(handle, kill=kill)

self.__init__()

def send_signal_to_process(self, signal, handle=None, group=False):

"""Sends the given ``signal`` to the specified process.

if os.sep == '\\':

raise RuntimeError('This keyword does not work on Windows.')

process = self._processes[handle]

signum = self._get_signal_number(signal)

logger.info('Sending signal %s (%d).' % (signal, signum))

if is_truthy(group) and hasattr(os, 'killpg'):

os.killpg(process.pid, signum)

elif hasattr(process, 'send_signal'):

process.send_signal(signum)

else:

raise RuntimeError('Sending signals is not supported '

'by this Python version.')

def _get_signal_number(self, int_or_name):

try:

return int(int_or_name)

except ValueError:

return self._convert_signal_name_to_number(int_or_name)

def _convert_signal_name_to_number(self, name):

try:

return getattr(signal_module,

name if name.startswith('SIG') else 'SIG' + name)

except AttributeError:

raise RuntimeError("Unsupported signal '%s'." % name)

def get_process_id(self, handle=None):

"""Returns the process ID (pid) of the process as an integer.

return self._processes[handle].pid

def get_process_object(self, handle=None):

"""Return the underlying ``subprocess.Popen`` object.

return self._processes[handle]

def get_process_result(self, handle=None, rc=False, stdout=False,

stderr=False, stdout_path=False, stderr_path=False):

"""Returns the specified `result object` or some of its attributes.

result = self._results[self._processes[handle]]

if result.rc is None:

raise RuntimeError('Getting results of unfinished processes '

'is not supported.')

attributes = self._get_result_attributes(result, rc, stdout, stderr,

stdout_path, stderr_path)

if not attributes:

return result

elif len(attributes) == 1:

return attributes[0]

return attributes

def _get_result_attributes(self, result, *includes):

attributes = (result.rc, result.stdout, result.stderr,

result.stdout_path, result.stderr_path)

includes = (is_truthy(incl) for incl in includes)

return tuple(attr for attr, incl in zip(attributes, includes) if incl)

def switch_process(self, handle):

"""Makes the specified process the current `active process`.

self._processes.switch(handle)

def _process_is_stopped(self, process, timeout):

stopped = lambda: process.poll() is not None

max_time = time.time() + timeout

while time.time() <= max_time and not stopped():

time.sleep(min(0.1, timeout))

return stopped()

def split_command_line(self, args, escaping=False):

"""Splits command line string into a list of arguments.

return cmdline2list(args, escaping=escaping)

def join_command_line(self, *args):

"""Joins arguments into one command line string.

if len(args) == 1 and is_list_like(args[0]):

args = args[0]

def __init__(self, process, stdout, stderr, stdin=None, rc=None,

output_encoding=None):

self._process = process

self.stdout_path = self._get_path(stdout)

self.stderr_path = self._get_path(stderr)

self.rc = rc

self._output_encoding = output_encoding

self._stdout = None

self._stderr = None

self._custom_streams = [stream for stream in (stdout, stderr, stdin)

if self._is_custom_stream(stream)]

def _get_path(self, stream):

return stream.name if self._is_custom_stream(stream) else None

def _is_custom_stream(self, stream):

return stream not in (subprocess.PIPE, subprocess.STDOUT, None)

@property

def stdout(self):

if self._stdout is None:

self._read_stdout()

return self._stdout

@property

def stderr(self):

if self._stderr is None:

self._read_stderr()

return self._stderr

def _read_stdout(self):

self._stdout = self._read_stream(self.stdout_path, self._process.stdout)

def _read_stderr(self):

self._stderr = self._read_stream(self.stderr_path, self._process.stderr)

def _read_stream(self, stream_path, stream):

if stream_path:

stream = open(stream_path, 'rb')

elif not self._is_open(stream):

return ''

try:

content = stream.read()

except IOError: # http://bugs.jython.org/issue2218

return ''

finally:

if stream_path:

stream.close()

return self._format_output(content)

def _is_open(self, stream):

return stream and not stream.closed

def _format_output(self, output):

output = console_decode(output, self._output_encoding)

output = output.replace('\r\n', '\n')

if output.endswith('\n'):

output = output[:-1]

return output

def close_streams(self):

standard_streams = self._get_and_read_standard_streams(self._process)

for stream in standard_streams + self._custom_streams:

if self._is_open(stream):

stream.close()

def _get_and_read_standard_streams(self, process):

stdin, stdout, stderr = process.stdin, process.stdout, process.stderr

if stdout:

self._read_stdout()

if stderr:

self._read_stderr()

return [stdin, stdout, stderr]

def __str__(self):

def __init__(self, cwd=None, shell=False, stdout=None, stderr=None, stdin='PIPE',

output_encoding='CONSOLE', alias=None, env=None, **rest):

self.cwd = self._get_cwd(cwd)

self.shell = is_truthy(shell)

self.alias = alias

self.output_encoding = output_encoding

self.stdout_stream = self._new_stream(stdout)

self.stderr_stream = self._get_stderr(stderr, stdout, self.stdout_stream)

self.stdin_stream = self._get_stdin(stdin)

self.env = self._construct_env(env, rest)

def _get_cwd(self, cwd):

if cwd:

return cwd.replace('/', os.sep)

return abspath('.')

def _new_stream(self, name):

if name == 'DEVNULL':

return open(os.devnull, 'w')

if name:

path = os.path.normpath(os.path.join(self.cwd, name))

return open(path, 'w')

return subprocess.PIPE

def _get_stderr(self, stderr, stdout, stdout_stream):

if stderr and stderr in ['STDOUT', stdout]:

if stdout_stream != subprocess.PIPE:

return stdout_stream

return subprocess.STDOUT

return self._new_stream(stderr)

def _get_stdin(self, stdin):

if not is_string(stdin):

return stdin

if stdin.upper() == 'NONE':

return None

if stdin == 'PIPE':

return subprocess.PIPE

path = os.path.normpath(os.path.join(self.cwd, stdin))

if os.path.isfile(path):

return open(path)

stdin_file = TemporaryFile()

if is_unicode(stdin):

stdin = console_encode(stdin, self.output_encoding, force=True)

stdin_file.write(stdin)

stdin_file.seek(0)

return stdin_file

def _construct_env(self, env, extra):

env = self._get_initial_env(env, extra)

if env is None:

return None

if WINDOWS:

env = NormalizedDict(env, spaceless=False)

self._add_to_env(env, extra)

if WINDOWS:

env = dict((key.upper(), env[key]) for key in env)

return env

def _get_initial_env(self, env, extra):

if env:

return dict((system_encode(k), system_encode(env[k])) for k in env)

if extra:

return os.environ.copy()

return None

def _add_to_env(self, env, extra):

for key in extra:

if not key.startswith('env:'):

raise RuntimeError("Keyword argument '%s' is not supported by "

"this keyword." % key)

env[system_encode(key[4:])] = system_encode(extra[key])

def get_command(self, command, arguments):

command = [system_encode(item) for item in [command] + arguments]

if not self.shell:

return command

if arguments:

return subprocess.list2cmdline(command)

return command[0]

@property

def popen_config(self):

config = {'stdout': self.stdout_stream,

'stderr': self.stderr_stream,

'stdin': self.stdin_stream,

'shell': self.shell,

'cwd': self.cwd,

'env': self.env}

# Close file descriptors regardless the Python version:

# https://github.com/robotframework/robotframework/issues/2794

if not WINDOWS:

config['close_fds'] = True

self._add_process_group_config(config)

return config

def _add_process_group_config(self, config):

if hasattr(os, 'setsid'):

config['preexec_fn'] = os.setsid

if hasattr(subprocess, 'CREATE_NEW_PROCESS_GROUP'):

config['creationflags'] = subprocess.CREATE_NEW_PROCESS_GROUP

@property

def result_config(self):

return {'stdout': self.stdout_stream,

'stderr': self.stderr_stream,

'stdin': self.stdin_stream,

'output_encoding': self.output_encoding}

def __str__(self):

return """\

env: %s""" % (self.cwd,

self.shell,

self._stream_name(self.stdout_stream),

self._stream_name(self.stderr_stream),

self._stream_name(self.stdin_stream),

self.alias,

self.env)

def _stream_name(self, stream):

if hasattr(stream, 'name'):

return stream.name

return {subprocess.PIPE: 'PIPE',

subprocess.STDOUT: 'STDOUT',