utils.py 文件源码

def run(self, timeout=None, is_json=False, **kwargs):
        """Run the self.command and wait up to given time period for results.

        :param timeout: how long to wait, in seconds, for the command to finish
        before terminating it
        :param is_json: hint whether output of the command is a JSON
        :return: triplet (return code, stdout, stderr), stdout will be a
        dictionary if `is_json` is True
        """
        logger.debug("running command '%s'; timeout '%s'", self.command, timeout)

        # this gets executed in a separate thread
        def target(**kwargs):
            try:
                self.process = Popen(self.command, universal_newlines=True, **kwargs)
                self.output, self.error = self.process.communicate()
                self.status = self.process.returncode
            except Exception:
                self.output = {} if is_json else []
                self.error = format_exc()
                self.status = -1

        # default stdout and stderr
        if 'stdout' not in kwargs:
            kwargs['stdout'] = PIPE
        if 'stderr' not in kwargs:
            kwargs['stderr'] = PIPE
        if 'update_env' in kwargs:
            # make sure we update environment, not override it
            kwargs['env'] = dict(os_environ, **kwargs['update_env'])
            kwargs.pop('update_env')

        # thread
        thread = Thread(target=target, kwargs=kwargs)
        thread.start()
        thread.join(timeout)

        # timeout reached, terminate the thread
        if thread.is_alive():
            logger.error('Command {cmd} timed out after {t} seconds'.format(cmd=self.command,
                                                                            t=timeout))
            # this is tricky - we need to make sure we kill the process with all its subprocesses;
            #  using just kill might create zombie process waiting for subprocesses to finish
            #  and leaving us hanging on thread.join()
            # TODO: we should do the same for get_command_output!
            killpg(getpgid(self.process.pid), signal.SIGKILL)
            thread.join()
            if not self.error:
                self.error = 'Killed by timeout after {t} seconds'.format(t=timeout)
        if self.output:
            if is_json:
                self.output = json.loads(self.output)
            else:
                self.output = [f for f in self.output.split('\n') if f]

        return self.status, self.output, self.error