python类SubprocessError()的实例源码

def send_message(self, message="", **kwargs):
        """Send a message to a command line."""
        try:
            proc = subprocess.Popen(self.command, universal_newlines=True,
                                    stdin=subprocess.PIPE, shell=True)
            proc.communicate(input=message)
            if proc.returncode != 0:
                _LOGGER.error('Command failed: %s', self.command)
        except subprocess.SubprocessError:
            _LOGGER.error('Error trying to exec Command: %s', self.command)

def run(self):
        for s in self.test_suites:
            script = os.path.join(s, 'run.py')
            start_time = time.time()
            manifest = self.load_manifest(s)
            os.chdir(s)

            info("Running tests from {0}".format(s))

            args = [e('${venvdir}/bin/python'), script]
            test = None
            try:
                test = subprocess.Popen(
                    args,
                    stdout=subprocess.PIPE,
                    stderr=subprocess.STDOUT,
                    close_fds=True
                )
                test.wait(timeout=manifest['timeout'])
            except subprocess.TimeoutExpired as err:
                self.generate_suite_error(
                    os.path.join(s, 'results.xml'),
                    manifest['name'],
                    time.time() - start_time,
                    'Test timeout reached',
                    err
                )
            except subprocess.SubprocessError as err:
                self.generate_suite_error(
                    os.path.join(s, 'results.xml'),
                    manifest['name'],
                    time.time() - start_time,
                    'Test could not be started',
                    err
                )

            out, err = test.communicate()

            if test and test.returncode:
                self.generate_suite_error(
                    os.path.join(s, 'results.xml'),
                    manifest['name'],
                    time.time() - start_time,
                    'Test process has returned an error',
                    out
                )

            info("{0} error:".format(script))
            print(out.decode('utf-8'))

def setup(hass, config):
    """Setup the shell_command component."""
    conf = config.get(DOMAIN, {})

    cache = {}

    def service_handler(call):
        """Execute a shell command service."""
        cmd = conf[call.service]

        if cmd in cache:
            prog, args, args_compiled = cache[cmd]
        elif ' ' not in cmd:
            prog = cmd
            args = None
            args_compiled = None
            cache[cmd] = prog, args, args_compiled
        else:
            prog, args = cmd.split(' ', 1)
            args_compiled = template.Template(args, hass)
            cache[cmd] = prog, args, args_compiled

        if args_compiled:
            try:
                rendered_args = args_compiled.render(call.data)
            except TemplateError as ex:
                _LOGGER.exception('Error rendering command template: %s', ex)
                return
        else:
            rendered_args = None

        if rendered_args == args:
            # no template used. default behavior
            shell = True
        else:
            # template used. Break into list and use shell=False for security
            cmd = [prog] + shlex.split(rendered_args)
            shell = False

        try:
            subprocess.call(cmd, shell=shell,
                            stdout=subprocess.DEVNULL,
                            stderr=subprocess.DEVNULL)
        except subprocess.SubprocessError:
            _LOGGER.exception('Error running command: %s', cmd)

    for name in conf.keys():
        hass.services.register(DOMAIN, name, service_handler)
    return True