Process.py

Go to the documentation of this file.

#  Copyright 2008-2015 Nokia Networks
#  Copyright 2016-     Robot Framework Foundation
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
 
import os
import signal as signal_module
import subprocess
import time
from tempfile import TemporaryFile
 
from robot.utils import (abspath, cmdline2list, ConnectionCache, console_decode,
                         console_encode, is_list_like, is_pathlike, is_string,
                         is_truthy, NormalizedDict, secs_to_timestr, system_decode,
                         system_encode, timestr_to_secs, WINDOWS)
from robot.version import get_version
from robot.api import logger
 
 
 
class Process:
    ROBOT_LIBRARY_SCOPE = 'GLOBAL'
    ROBOT_LIBRARY_VERSION = get_version()
    TERMINATE_TIMEOUT = 30
    KILL_TIMEOUT = 10
 
    def __init__(self):
        self._processes_processes = ConnectionCache('No active process.')
        self._results_results = {}
 
    
    def run_process(self, command, *arguments, **configuration):
        current = self._processes_processes.current
        timeout = configuration.pop('timeout', None)
        on_timeout = configuration.pop('on_timeout', 'terminate')
        try:
            handle = self.start_processstart_process(command, *arguments, **configuration)
            return self.wait_for_processwait_for_process(handle, timeout, on_timeout)
        finally:
            self._processes_processes.current = current
 
    
    def start_process(self, command, *arguments, **configuration):
        conf = ProcessConfiguration(**configuration)
        command = conf.get_command(command, list(arguments))
        self._log_start_log_start(command, conf)
        process = subprocess.Popen(command, **conf.popen_config)
        self._results_results[process] = ExecutionResult(process, **conf.result_config)
        self._processes_processes.register(process, alias=conf.alias)
        return self._processes_processes.current
 
    def _log_start(self, command, config):
        if is_list_like(command):
            command = self.join_command_linejoin_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):
        return self._processes_processes[handle].poll() is None
 
    
    def process_should_be_running(self, handle=None,
                                  error_message='Process is not running.'):
        if not self.is_process_runningis_process_running(handle):
            raise AssertionError(error_message)
 
    
    def process_should_be_stopped(self, handle=None,
                                  error_message='Process is running.'):
        if self.is_process_runningis_process_running(handle):
            raise AssertionError(error_message)
 
    
    def wait_for_process(self, handle=None, timeout=None, on_timeout='continue'):
        process = self._processes_processes[handle]
        logger.info('Waiting for process to complete.')
        timeout = self._get_timeout_get_timeout(timeout)
        if timeout > 0:
            if not self._process_is_stopped_process_is_stopped(process, timeout):
                logger.info('Process did not complete in %s.'
                            % secs_to_timestr(timeout))
                return self._manage_process_timeout_manage_process_timeout(handle, on_timeout.lower())
        return self._wait_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_processterminate_process(handle)
        elif on_timeout == 'kill':
            return self.terminate_processterminate_process(handle, kill=True)
        else:
            logger.info('Leaving process intact.')
            return None
 
    def _wait(self, process):
        result = self._results_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):
        process = self._processes_processes[handle]
        if not hasattr(process, 'terminate'):
            raise RuntimeError('Terminating processes is not supported '
                               'by this Python version.')
        terminator = self._kill_kill if is_truthy(kill) else self._terminate_terminate
        try:
            terminator(process)
        except OSError:
            if not self._process_is_stopped_process_is_stopped(process, self.KILL_TIMEOUTKILL_TIMEOUT):
                raise
            logger.debug('Ignored OSError because process was stopped.')
        return self._wait_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_is_stopped(process, self.KILL_TIMEOUTKILL_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_is_stopped(process, self.TERMINATE_TIMEOUTTERMINATE_TIMEOUT):
            logger.info('Graceful termination failed.')
            self._kill_kill(process)
 
    
    def terminate_all_processes(self, kill=False):
        for handle in range(1, len(self._processes_processes) + 1):
            if self.is_process_runningis_process_running(handle):
                self.terminate_processterminate_process(handle, kill=kill)
        self.__init____init__()
 
    
    def send_signal_to_process(self, signal, handle=None, group=False):
        if os.sep == '\\':
            raise RuntimeError('This keyword does not work on Windows.')
        process = self._processes_processes[handle]
        signum = self._get_signal_number_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_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):
        return self._processes_processes[handle].pid
 
    
    def get_process_object(self, handle=None):
        return self._processes_processes[handle]
 
    
    def get_process_result(self, handle=None, rc=False, stdout=False,
                           stderr=False, stdout_path=False, stderr_path=False):
        result = self._results_results[self._processes_processes[handle]]
        if result.rc is None:
            raise RuntimeError('Getting results of unfinished processes '
                               'is not supported.')
        attributes = self._get_result_attributes_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):
        self._processes_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):
        return cmdline2list(args, escaping=escaping)
 
    
    def join_command_line(self, *args):
        if len(args) == 1 and is_list_like(args[0]):
            args = args[0]
        return subprocess.list2cmdline(args)
 
 
class ExecutionResult:
 
    def __init__(self, process, stdout, stderr, stdin=None, rc=None,
                 output_encoding=None):
        self._process_process = process
        self.stdout_pathstdout_path = self._get_path_get_path(stdout)
        self.stderr_pathstderr_path = self._get_path_get_path(stderr)
        self.rcrc = rc
        self._output_encoding_output_encoding = output_encoding
        self._stdout_stdout = None
        self._stderr_stderr = None
        self._custom_streams_custom_streams = [stream for stream in (stdout, stderr, stdin)
                                if self._is_custom_stream_is_custom_stream(stream)]
 
    def _get_path(self, stream):
        return stream.name if self._is_custom_stream_is_custom_stream(stream) else None
 
    def _is_custom_stream(self, stream):
        return stream not in (subprocess.PIPE, subprocess.STDOUT, None)
 
    @property
    stdout = property
    
    def stdout(self):
        if self._stdout_stdout is None:
            self._read_stdout_read_stdout()
        return self._stdout_stdout
 
    @property
    stderr = property
    
    def stderr(self):
        if self._stderr_stderr is None:
            self._read_stderr_read_stderr()
        return self._stderr_stderr
 
    def _read_stdout(self):
        self._stdout_stdout = self._read_stream_read_stream(self.stdout_pathstdout_path, self._process_process.stdout)
 
    def _read_stderr(self):
        self._stderr_stderr = self._read_stream_read_stream(self.stderr_pathstderr_path, self._process_process.stderr)
 
    def _read_stream(self, stream_path, stream):
        if stream_path:
            stream = open(stream_path, 'rb')
        elif not self._is_open_is_open(stream):
            return ''
        try:
            content = stream.read()
        except IOError:
            content = ''
        finally:
            if stream_path:
                stream.close()
        return self._format_output_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_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_get_and_read_standard_streams(self._process_process)
        for stream in standard_streams + self._custom_streams_custom_streams:
            if self._is_open_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_read_stdout()
        if stderr:
            self._read_stderr_read_stderr()
        return [stdin, stdout, stderr]
 
    def __str__(self):
        return '<result object with rc %d>' % self.rcrc
 
 
class ProcessConfiguration:
 
    def __init__(self, cwd=None, shell=False, stdout=None, stderr=None, stdin='PIPE',
                 output_encoding='CONSOLE', alias=None, env=None, **rest):
        self.cwdcwd = os.path.normpath(cwd) if cwd else abspath('.')
        self.shellshell = is_truthy(shell)
        self.aliasalias = alias
        self.output_encodingoutput_encoding = output_encoding
        self.stdout_streamstdout_stream = self._new_stream_new_stream(stdout)
        self.stderr_streamstderr_stream = self._get_stderr_get_stderr(stderr, stdout, self.stdout_streamstdout_stream)
        self.stdin_streamstdin_stream = self._get_stdin_get_stdin(stdin)
        self.envenv = self._construct_env_construct_env(env, rest)
 
    def _new_stream(self, name):
        if name == 'DEVNULL':
            return open(os.devnull, 'w')
        if name:
            path = os.path.normpath(os.path.join(self.cwdcwd, 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_new_stream(stderr)
 
    def _get_stdin(self, stdin):
        if is_pathlike(stdin):
            stdin = str(stdin)
        elif not is_string(stdin):
            return stdin
        elif stdin.upper() == 'NONE':
            return None
        elif stdin == 'PIPE':
            return subprocess.PIPE
        path = os.path.normpath(os.path.join(self.cwdcwd, stdin))
        if os.path.isfile(path):
            return open(path)
        stdin_file = TemporaryFile()
        stdin_file.write(console_encode(stdin, self.output_encodingoutput_encoding, force=True))
        stdin_file.seek(0)
        return stdin_file
 
    def _construct_env(self, env, extra):
        env = self._get_initial_env_get_initial_env(env, extra)
        if env is None:
            return None
        if WINDOWS:
            env = NormalizedDict(env, spaceless=False)
        self._add_to_env_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.shellshell:
            return command
        if arguments:
            return subprocess.list2cmdline(command)
        return command[0]
 
    @property
    popen_config = property
    
    def popen_config(self):
        config = {'stdout': self.stdout_streamstdout_stream,
                  'stderr': self.stderr_streamstderr_stream,
                  'stdin': self.stdin_streamstdin_stream,
                  'shell': self.shellshell,
                  'cwd': self.cwdcwd,
                  'env': self.envenv}
        # 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_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
    result_config = property
    
    def result_config(self):
        return {'stdout': self.stdout_streamstdout_stream,
                'stderr': self.stderr_streamstderr_stream,
                'stdin': self.stdin_streamstdin_stream,
                'output_encoding': self.output_encodingoutput_encoding}
 
    def __str__(self):
        return """\
cwd:     %s
shell:   %s
stdout:  %s
stderr:  %s
stdin:   %s
alias:   %s
env:     %s""" % (self.cwdcwd,
                  self.shellshell,
                  self._stream_name_stream_name(self.stdout_streamstdout_stream),
                  self._stream_name_stream_name(self.stderr_streamstderr_stream),
                  self._stream_name_stream_name(self.stdin_streamstdin_stream),
                  self.aliasalias,
                  self.envenv)
 
    def _stream_name(self, stream):
        if hasattr(stream, 'name'):
            return stream.name
        return {subprocess.PIPE: 'PIPE',
                subprocess.STDOUT: 'STDOUT',
                None: 'None'}.get(stream, stream)