Python asyncio 模块，create_subprocess_exec() 实例源码

def execute_process(*cmd, log=None, loop=None):
    '''
    Wrapper around asyncio.create_subprocess_exec.

    '''
    p = await asyncio.create_subprocess_exec(
        *cmd,
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        loop=loop)
    stdout, stderr = await p.communicate()
    if log:
        log.debug("Exec %s -> %d", cmd, p.returncode)
        if stdout:
            log.debug(stdout.decode('utf-8'))
        if stderr:
            log.debug(stderr.decode('utf-8'))
    return p.returncode == 0

def _scp(self, source, destination, scp_opts):
        """ Execute an scp command. Requires a fully qualified source and
        destination.
        """
        cmd = [
            'scp',
            '-i', os.path.expanduser('~/.local/share/juju/ssh/juju_id_rsa'),
            '-o', 'StrictHostKeyChecking=no',
            '-q',
            '-B',
            source, destination
        ]
        cmd += scp_opts.split()
        loop = self.model.loop
        process = await asyncio.create_subprocess_exec(*cmd, loop=loop)
        await process.wait()
        if process.returncode != 0:
            raise JujuError("command failed: %s" % cmd)

def run_df():
    print('in run_df')

    buffer = bytearray()

    create = asyncio.create_subprocess_exec(
        'df', '-hl',
        stdout=asyncio.subprocess.PIPE,
    )
    print('launching process')
    proc = await create
    print('process started {}'.format(proc.pid))

    while True:
        line = await proc.stdout.readline()
        print('read {!r}'.format(line))
        if not line:
            print('no more output from command')
            break
        buffer.extend(line)

    print('waiting for process to complete')
    await proc.wait()

    return_code = proc.returncode
    print('return code {}'.format(return_code))
    if not return_code:
        cmd_output = bytes(buffer).decode()
        results = _parse_results(cmd_output)
    else:
        results = []

    return (return_code, results)

def task():
    rfd, wfd = os.pipe()
    args = [sys.executable, '-c', code, str(rfd)]
    proc = yield from asyncio.create_subprocess_exec(
                          *args,
                          pass_fds={rfd},
                          stdout=subprocess.PIPE)

    pipe = open(wfd, 'wb', 0)
    transport, _ = yield from loop.connect_write_pipe(asyncio.Protocol,
                                                      pipe)
    transport.write(b'data')

    stdout, stderr = yield from proc.communicate()
    print("stdout = %r" % stdout.decode())
    transport.close()

def test_communicate(self):
        args = PROGRAM_CAT

        @asyncio.coroutine
        def run(data):
            proc = yield from asyncio.create_subprocess_exec(
                                          *args,
                                          stdin=subprocess.PIPE,
                                          stdout=subprocess.PIPE,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(data)
            return proc.returncode, stdout

        task = run(b'some data')
        task = asyncio.wait_for(task, 60.0, loop=self.loop)
        exitcode, stdout = self.loop.run_until_complete(task)
        self.assertEqual(exitcode, 0)
        self.assertEqual(stdout, b'some data')

def test_send_signal(self):
        code = 'import time; print("sleeping", flush=True); time.sleep(3600)'
        args = [sys.executable, '-c', code]
        create = asyncio.create_subprocess_exec(*args,
                                                stdout=subprocess.PIPE,
                                                loop=self.loop)
        proc = self.loop.run_until_complete(create)

        @asyncio.coroutine
        def send_signal(proc):
            # basic synchronization to wait until the program is sleeping
            line = yield from proc.stdout.readline()
            self.assertEqual(line, b'sleeping\n')

            proc.send_signal(signal.SIGHUP)
            returncode = (yield from proc.wait())
            return returncode

        returncode = self.loop.run_until_complete(send_signal(proc))
        self.assertEqual(-signal.SIGHUP, returncode)

def test_stdin_not_inheritable(self):
        # Tulip issue #209: stdin must not be inheritable, otherwise
        # the Process.communicate() hangs
        @asyncio.coroutine
        def len_message(message):
            code = 'import sys; data = sys.stdin.read(); print(len(data))'
            proc = yield from asyncio.create_subprocess_exec(
                                          sys.executable, '-c', code,
                                          stdin=asyncio.subprocess.PIPE,
                                          stdout=asyncio.subprocess.PIPE,
                                          stderr=asyncio.subprocess.PIPE,
                                          close_fds=False,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(message)
            exitcode = yield from proc.wait()
            return (stdout, exitcode)

        output, exitcode = self.loop.run_until_complete(len_message(b'abc'))
        self.assertEqual(output.rstrip(), b'3')
        self.assertEqual(exitcode, 0)

def test_cancel_process_wait(self):
        # Issue #23140: cancel Process.wait()

        @asyncio.coroutine
        def cancel_wait():
            proc = yield from asyncio.create_subprocess_exec(
                                          *PROGRAM_BLOCKED,
                                          loop=self.loop)

            # Create an internal future waiting on the process exit
            task = self.loop.create_task(proc.wait())
            self.loop.call_soon(task.cancel)
            try:
                yield from task
            except asyncio.CancelledError:
                pass

            # Cancel the future
            task.cancel()

            # Kill the process and wait until it is done
            proc.kill()
            yield from proc.wait()

        self.loop.run_until_complete(cancel_wait())

def test_cancel_make_subprocess_transport_exec(self):
        @asyncio.coroutine
        def cancel_make_transport():
            coro = asyncio.create_subprocess_exec(*PROGRAM_BLOCKED,
                                                  loop=self.loop)
            task = self.loop.create_task(coro)

            self.loop.call_soon(task.cancel)
            try:
                yield from task
            except asyncio.CancelledError:
                pass

        # ignore the log:
        # "Exception during subprocess creation, kill the subprocess"
        with test_utils.disable_logger():
            self.loop.run_until_complete(cancel_make_transport())

def test_pipe_to_log(context, event_loop):
    cmd = r""">&2 echo "foo" && echo "bar" && exit 0"""
    proc = event_loop.run_until_complete(
        asyncio.create_subprocess_exec(
            "bash", "-c", cmd,
            stdout=PIPE, stderr=PIPE, stdin=None
        )
    )
    tasks = []
    with swlog.get_log_filehandle(context) as log_fh:
        tasks.append(swlog.pipe_to_log(proc.stderr, filehandles=[log_fh]))
        tasks.append(swlog.pipe_to_log(proc.stdout, filehandles=[log_fh]))
        event_loop.run_until_complete(asyncio.wait(tasks))
        event_loop.run_until_complete(proc.wait())
    log_file = swlog.get_log_filename(context)
    assert read(log_file) in ("foo\nbar\n", "bar\nfoo\n")

def launch_second_instances():
    temp_dir = sys.argv[1]
    if not os.path.exists(temp_dir):
        os.makedirs(temp_dir)
    job1 = subprocess.Popen(
        [sys.executable, __file__,
         os.path.join(temp_dir, "one"),
         os.path.join(temp_dir, "two"),
         os.path.join(temp_dir, "three")],
    )
    loop = asyncio.get_event_loop()
    job2 = asyncio.create_subprocess_exec(
        sys.executable,
        __file__,
        os.path.join(temp_dir, "four"),
        os.path.join(temp_dir, "five"),
        os.path.join(temp_dir, "six"),
    )
    loop.run_until_complete(job2)
    job1.wait()

def get_latest_tag(path,
                         exec_function=asyncio.create_subprocess_exec):
    """Get the latest tag in path.

    Args:
        path (str): the path to run ``git describe --abbrev=0`` in.

    Returns:
        str: the tag name found.

    Raises:
        ScriptWorkerRetryException: on failure.

    """
    proc = await exec_function(
        'git', "describe", "--abbrev=0", cwd=path,
        stdout=PIPE, stderr=DEVNULL, stdin=DEVNULL, close_fds=True,
    )
    tag, err = await proc.communicate()
    exitcode = await proc.wait()
    if exitcode:
        raise ScriptWorkerRetryException(
            "Can't get tag at {}: {}!".format(path, err)
        )
    return tag.decode('utf-8').rstrip()

def _start_redis_server(loop):
    print('Running Redis server REDIS_HOST=%r REDIS_PORT=%r...' % (HOST, PORT))

    redis_srv = loop.run_until_complete(
        asyncio.create_subprocess_exec(
            'redis-server',
            '--port', str(PORT),
            ('--bind' if PORT else '--unixsocket'), HOST,
            '--maxclients', '100',
            '--save', '""',
            '--loglevel', 'warning',
            loop=loop,
            stdout=asyncio.subprocess.DEVNULL,
            stderr=asyncio.subprocess.DEVNULL))
    loop.run_until_complete(asyncio.sleep(.05, loop=loop))
    return redis_srv

def image_convert_to_png(image):
    path_imagemagick = _externals["bot"].config.get_option("image.imagemagick") or "/usr/bin/convert"
    cmd = (path_imagemagick, "-", "png:-")

    try:
        proc = await asyncio.create_subprocess_exec(
            *cmd,
            stdin = PIPE,
            stdout = PIPE,
            stderr = PIPE )

        (stdout_data, stderr_data) = await proc.communicate(input=image)

        return stdout_data

    except FileNotFoundError:
        logger.error("imagemagick not found at path {}".format(path_imagemagick))
        return False

def run_cmd(self, *args, **kwargs):
        if len(args) > 1:
            value = args
        elif args:
            value = args[0]
        elif kwargs:
            value = kwargs
        else:
            value = None
        cmd = self.make_command(value)
        self.logger.info(' '.join(cmd))
        if len(cmd) > 1:
            coro = asyncio.create_subprocess_exec
        else:
            coro = asyncio.create_subprocess_shell
        self._process = await coro(
            *cmd, stdout=self._stdout, stderr=self._stderr, loop=self.loop)
        if self._wait:
            await self._process.wait()
        if self._stdout:
            data = await self._process.stdout.read()
            return self.decode(data)

def convert_file(fname, format, outdir=None): 
    if not outdir:
        outdir=os.path.dirname(fname)
        out_file = os.path.splitext(fname)[0] +'.'+format
    else:
        out_file=os.path.join(outdir, os.path.splitext(os.path.basename(fname))[0]+'.'+format)

    async with OOEnvironment() as env:
        cmd=(OOFFICE, '--headless', '-env:UserInstallation=file://%s'%env, '--convert-to', format,
             '--outdir', outdir, fname)
        proc = await asyncio.create_subprocess_exec(*cmd)
        try:
            return_code = await asyncio.wait_for(proc.wait(), 240)
        except asyncio.TimeoutError:
            proc.terminate()
            try:
                await asyncio.wait_for(proc.wait(), 10)
            except asyncio.TimeoutError:
                proc.kill()
    #older version of LibreOffice are returning non return code even if file is created
    if await aos.path.exists(out_file):
        return out_file
    else:
        logger.error('Failed %s with code %d',' '.join(cmd), return_code)

def test_mock_subprocess(event_loop):

    async def run(stdin, stdout, stderr):
        message = await asyncio.wait_for(stdin.read(), timeout=5.0)
        assert message == b'FUBAR'
        stdout.write(b'THIS IS')
        stderr.write(b'SPARTA!')
        return 456

    with mock_subprocess(run):
        process = await asyncio.create_subprocess_exec(
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            loop=event_loop,
        )
        assert process.pid != 0
        assert process.returncode is None
        stdout, stderr = await asyncio.wait_for(
            process.communicate(input=b'FUBAR'), timeout=5.0
        )
        assert process.returncode == 456
        assert stdout == b'THIS IS'
        assert stderr == b'SPARTA!'

def test_mock_subprocess_argv_defaults(event_loop):

    async def run(argv):
        assert argv == []

    with mock_subprocess(run):
        process = await asyncio.create_subprocess_exec(
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            loop=event_loop,
        )
        assert process.pid != 0
        assert process.returncode is None
        status = await asyncio.wait_for(process.wait(), timeout=5.0)
        assert status == 0
        assert process.returncode == 0

def test_mock_subprocess_argv(event_loop):

    async def run(argv):
        assert argv == ['a', 'b', 'c']

    with mock_subprocess(run):
        process = await asyncio.create_subprocess_exec(
            'a', 'b', 'c',
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            loop=event_loop,
        )
        assert process.pid != 0
        assert process.returncode is None
        status = await asyncio.wait_for(process.wait(), timeout=5.0)
        assert status == 0
        assert process.returncode == 0

def test_mock_subprocess_env_default(event_loop):

    async def run(env):
        assert env == os.environ
        assert env is not os.environ

    with mock_subprocess(run):
        process = await asyncio.create_subprocess_exec(
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            loop=event_loop,
        )
        assert process.pid != 0
        assert process.returncode is None
        status = await asyncio.wait_for(process.wait(), timeout=5.0)
        assert status == 0
        assert process.returncode == 0

def test_mock_subprocess_env(event_loop):

    async def run(env):
        assert env == {
            'A': 'B',
            'C': 'D',
        }

    with mock_subprocess(run):
        process = await asyncio.create_subprocess_exec(
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            env={
                'A': 'B',
                'C': 'D',
            },
            loop=event_loop,
        )
        assert process.pid != 0
        assert process.returncode is None
        status = await asyncio.wait_for(process.wait(), timeout=5.0)
        assert status == 0
        assert process.returncode == 0

def test_mock_subprocess_terminate(event_loop):

    async def run():
        await asyncio.wait_for(asyncio.Future(), timeout=5.0)

    with mock_subprocess(run):
        process = await asyncio.create_subprocess_exec(
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
            creationflags=512,  # subprocess.CREATE_NEW_PROCESS_GROUP
            loop=event_loop,
        )
        assert process.pid != 0
        assert process.returncode is None
        process.terminate()
        stdout, stderr = await asyncio.wait_for(
            process.communicate(), timeout=5.0
        )
        assert process.returncode == 1
        assert stdout == b''
        assert stderr == b''

def test_mock_subprocess_kill_posix(platform, event_loop):

    async def run(signals):
        sig = await asyncio.wait_for(signals.get(), timeout=5.0)
        assert sig == signal.SIGINT

    with mock.patch('sys.platform', platform):
        with mock_subprocess(run):
            process = await asyncio.create_subprocess_exec(
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                creationflags=512,  # subprocess.CREATE_NEW_PROCESS_GROUP
                loop=event_loop,
            )
            assert process.pid != 0
            assert process.returncode is None
            process.kill()
            stdout, stderr = await asyncio.wait_for(
                process.communicate(), timeout=5.0
            )
            assert process.returncode == 0
            assert stdout == b''
            assert stderr == b''

def test_mock_subprocess_kill_win32(event_loop):

    async def run():
        await asyncio.wait_for(asyncio.Future(), timeout=5.0)

    with mock.patch('sys.platform', 'win32'):
        with mock_subprocess(run):
            process = await asyncio.create_subprocess_exec(
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                loop=event_loop,
            )
            assert process.pid != 0
            assert process.returncode is None
            process.kill()
            stdout, stderr = await asyncio.wait_for(
                process.communicate(), timeout=5.0
            )
            assert process.returncode == 1
            assert stdout == b''
            assert stderr == b''

def test_communicate(self):
        args = PROGRAM_CAT

        @asyncio.coroutine
        def run(data):
            proc = yield from asyncio.create_subprocess_exec(
                                          *args,
                                          stdin=subprocess.PIPE,
                                          stdout=subprocess.PIPE,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(data)
            return proc.returncode, stdout

        task = run(b'some data')
        task = asyncio.wait_for(task, 60.0, loop=self.loop)
        exitcode, stdout = self.loop.run_until_complete(task)
        self.assertEqual(exitcode, 0)
        self.assertEqual(stdout, b'some data')

def test_stdin_not_inheritable(self):
        # asyncio issue #209: stdin must not be inheritable, otherwise
        # the Process.communicate() hangs
        @asyncio.coroutine
        def len_message(message):
            code = 'import sys; data = sys.stdin.read(); print(len(data))'
            proc = yield from asyncio.create_subprocess_exec(
                                          sys.executable, '-c', code,
                                          stdin=asyncio.subprocess.PIPE,
                                          stdout=asyncio.subprocess.PIPE,
                                          stderr=asyncio.subprocess.PIPE,
                                          close_fds=False,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(message)
            exitcode = yield from proc.wait()
            return (stdout, exitcode)

        output, exitcode = self.loop.run_until_complete(len_message(b'abc'))
        self.assertEqual(output.rstrip(), b'3')
        self.assertEqual(exitcode, 0)

def test_cancel_process_wait(self):
        # Issue #23140: cancel Process.wait()

        @asyncio.coroutine
        def cancel_wait():
            proc = yield from asyncio.create_subprocess_exec(
                                          *PROGRAM_BLOCKED,
                                          loop=self.loop)

            # Create an internal future waiting on the process exit
            task = self.loop.create_task(proc.wait())
            self.loop.call_soon(task.cancel)
            try:
                yield from task
            except asyncio.CancelledError:
                pass

            # Cancel the future
            task.cancel()

            # Kill the process and wait until it is done
            proc.kill()
            yield from proc.wait()

        self.loop.run_until_complete(cancel_wait())

def test_cancel_make_subprocess_transport_exec(self):
        @asyncio.coroutine
        def cancel_make_transport():
            coro = asyncio.create_subprocess_exec(*PROGRAM_BLOCKED,
                                                  loop=self.loop)
            task = self.loop.create_task(coro)

            self.loop.call_soon(task.cancel)
            try:
                yield from task
            except asyncio.CancelledError:
                pass

        # ignore the log:
        # "Exception during subprocess creation, kill the subprocess"
        with test_utils.disable_logger():
            self.loop.run_until_complete(cancel_make_transport())

def start_gui_for_stubdomain(self, vm, force=False):
        '''Start GUI daemon (qubes-guid) connected to a stubdomain

        This function is a coroutine.
        '''
        want_stubdom = force
        # if no 'gui' feature set at all, assume no gui agent installed
        if not want_stubdom and \
                vm.features.check_with_template('gui', None) is None:
            want_stubdom = True
        if not want_stubdom and vm.debug:
            want_stubdom = True
        if not want_stubdom:
            return
        if os.path.exists(self.guid_pidfile(vm.stubdom_xid)):
            return
        vm.log.info('Starting GUI (stubdomain)')
        guid_cmd = self.common_guid_args(vm)
        guid_cmd.extend(['-d', str(vm.stubdom_xid), '-t', str(vm.xid)])

        yield from asyncio.create_subprocess_exec(*guid_cmd)

def test_communicate(self):
        args = PROGRAM_CAT

        @asyncio.coroutine
        def run(data):
            proc = yield from asyncio.create_subprocess_exec(
                                          *args,
                                          stdin=subprocess.PIPE,
                                          stdout=subprocess.PIPE,
                                          loop=self.loop)
            stdout, stderr = yield from proc.communicate(data)
            return proc.returncode, stdout

        task = run(b'some data')
        task = asyncio.wait_for(task, 60.0, loop=self.loop)
        exitcode, stdout = self.loop.run_until_complete(task)
        self.assertEqual(exitcode, 0)
        self.assertEqual(stdout, b'some data')

def call(cls, rule):
            while True:
                # All our rules are in the "nat" table
                rule = ('-t', 'nat') + rule
                creator = asyncio.create_subprocess_exec(cls.IPTABLES, *rule, stderr=asyncio.subprocess.PIPE)
                proc = await creator
                # Wait for exit and get exit code
                res = await proc.wait()
                # Read error messages
                err = await proc.stderr.readline()
                # Handle errors
                if res:
                    # On lock (res == 4) or resource error we need to try again
                    if res == 4 or err == b"iptables: Resource temporarily unavailable.\n":
                        continue
                    # Expected errors
                    if res != 1 or (err != b"iptables: No chain/target/match by that name.\n" and
                                    err != b"iptables: Chain already exists.\n"):
                        error("Error (" + str(res) + "): " + err.rstrip().decode())
                # We are done
                break
            return res

def convert_pdf_to_jpg(user_path, the_hash) -> str:
    tex_logger.debug("Converting to jpg...")
    os.chdir(user_path)

    conversion_process = await asyncio.create_subprocess_exec(*["gs", "-o", "{}.jpg".format(the_hash), "-sDEVICE=jpeg",
                                                                "-dJPEGQ=100", "-r1000", "the_latex.pdf"],
                                                              stdout=asyncio.subprocess.DEVNULL)
    try:
        await conversion_process.wait()
    except asyncio.CancelledError:
        tex_logger.debug("Conversion cancelled.\n")
        conversion_process.kill()
        raise

    os.chdir(os.path.dirname(user_path))
    tex_logger.debug(" Converted.\n")
    return os.path.join(user_path, "{}.jpg".format(the_hash))

def echo(msg):
    # Run an echo subprocess
    process = await asyncio.create_subprocess_exec(
        'cat',
        # stdin must a pipe to be accessible as process.stdin
        stdin=asyncio.subprocess.PIPE,
        # stdout must a pipe to be accessible as process.stdout
        stdout=asyncio.subprocess.PIPE)
    # Write message
    print('Writing {!r} ...'.format(msg))
    process.stdin.write(msg.encode() + b'\n')
    # Read reply
    data = await process.stdout.readline()
    reply = data.decode().strip()
    print('Received {!r}'.format(reply))
    # Stop the subprocess
    process.terminate()
    code = await process.wait()
    print('Terminated with code {}'.format(code))

def runner(argv, timeout=0):
    """
    Run the input command-line executable (specified in a Popen-style list) and
    return its exit code. Optionally specify a timeout. If timeout is 0 or
    None, simply wait until the process is done.
    """

    def stringify(xs):
        return map(str, xs)

    argv = list(stringify(argv))
    proc = await asyncio.create_subprocess_exec(*argv,
                                                stdout=asyncio.subprocess.PIPE,
                                                stderr=asyncio.subprocess.PIPE)
    if timeout <= 0:
        timeout = None
    await asyncio.wait_for(proc.wait(), timeout=timeout)
    return proc.returncode

def atm_process(event_loop):
    process = await asyncio.create_subprocess_exec(
        sys.executable, b"../atm.py", loop=event_loop
    )
    await asyncio.sleep(0.5)
    yield process
    process.terminate()
    await process.wait()

def to_upper(input):
    print('in to_upper')

    create = asyncio.create_subprocess_exec(
        'tr', '[:lower:]', '[:upper:]',
        stdout=asyncio.subprocess.PIPE,
        stdin=asyncio.subprocess.PIPE,
    )
    print('launching process')
    proc = await create
    print('pid {}'.format(proc.pid))

    print('communicating with process')
    stdout, stderr = await proc.communicate(input.encode())

    print('waiting for process to complete')
    await proc.wait()

    return_code = proc.returncode
    print('return code {}'.format(return_code))
    if not return_code:
        results = bytes(stdout).decode()
    else:
        results = ''

    return (return_code, results)

def ls(loop):
    proc = yield from asyncio.create_subprocess_exec("ls",
                                                     stdout=PIPE)
    while True:
        line = yield from proc.stdout.readline()
        if not line:
            break
        print("ls>>", line.decode('ascii').rstrip())
    try:
        proc.send_signal(signal.SIGINT)
    except ProcessLookupError:
        pass

def test_call(*args, timeout=None):
    proc = yield from asyncio.create_subprocess_exec(*args)
    try:
        exitcode = yield from asyncio.wait_for(proc.wait(), timeout)
        print("%s: exit code %s" % (' '.join(args), exitcode))
    except asyncio.TimeoutError:
        print("timeout! (%.1f sec)" % timeout)
        proc.kill()
        yield from proc.wait()


#
# ??:
#

def task():
    rfd, wfd = os.pipe()
    args = [sys.executable, '-c', code, str(wfd)]

    pipe = open(rfd, 'rb', 0)
    reader = asyncio.StreamReader(loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    transport, _ = yield from loop.connect_read_pipe(lambda: protocol, pipe)

    proc = yield from asyncio.create_subprocess_exec(*args, pass_fds={wfd})
    yield from proc.wait()

    os.close(wfd)
    data = yield from reader.read()
    print("read = %r" % data.decode())

def test_read_all_from_pipe_reader(self):
        # See Tulip issue 168.  This test is derived from the example
        # subprocess_attach_read_pipe.py, but we configure the
        # StreamReader's limit so that twice it is less than the size
        # of the data writter.  Also we must explicitly attach a child
        # watcher to the event loop.

        code = """\
import os, sys
fd = int(sys.argv[1])
os.write(fd, b'data')
os.close(fd)
"""
        rfd, wfd = os.pipe()
        args = [sys.executable, '-c', code, str(wfd)]

        pipe = open(rfd, 'rb', 0)
        reader = asyncio.StreamReader(loop=self.loop, limit=1)
        protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop)
        transport, _ = self.loop.run_until_complete(
            self.loop.connect_read_pipe(lambda: protocol, pipe))

        watcher = asyncio.SafeChildWatcher()
        watcher.attach_loop(self.loop)
        try:
            asyncio.set_child_watcher(watcher)
            create = asyncio.create_subprocess_exec(*args,
                                                    pass_fds={wfd},
                                                    loop=self.loop)
            proc = self.loop.run_until_complete(create)
            self.loop.run_until_complete(proc.wait())
        finally:
            asyncio.set_child_watcher(None)

        os.close(wfd)
        data = self.loop.run_until_complete(reader.read(-1))
        self.assertEqual(data, b'data')

def test_kill(self):
        args = PROGRAM_BLOCKED
        create = asyncio.create_subprocess_exec(*args, loop=self.loop)
        proc = self.loop.run_until_complete(create)
        proc.kill()
        returncode = self.loop.run_until_complete(proc.wait())
        if sys.platform == 'win32':
            self.assertIsInstance(returncode, int)
            # expect 1 but sometimes get 0
        else:
            self.assertEqual(-signal.SIGKILL, returncode)

def test_terminate(self):
        args = PROGRAM_BLOCKED
        create = asyncio.create_subprocess_exec(*args, loop=self.loop)
        proc = self.loop.run_until_complete(create)
        proc.terminate()
        returncode = self.loop.run_until_complete(proc.wait())
        if sys.platform == 'win32':
            self.assertIsInstance(returncode, int)
            # expect 1 but sometimes get 0
        else:
            self.assertEqual(-signal.SIGTERM, returncode)

def prepare_broken_pipe_test(self):
        # buffer large enough to feed the whole pipe buffer
        large_data = b'x' * support.PIPE_MAX_SIZE

        # the program ends before the stdin can be feeded
        create = asyncio.create_subprocess_exec(
                             sys.executable, '-c', 'pass',
                             stdin=subprocess.PIPE,
                             loop=self.loop)
        proc = self.loop.run_until_complete(create)
        return (proc, large_data)

def open(self, cmd, input_source, output="-", extra_cmd=None,
             stdout_pipe=True, stderr_pipe=False):
        """Start a ffmpeg instance and pipe output."""
        stdout = asyncio.subprocess.PIPE if stdout_pipe\
            else asyncio.subprocess.DEVNULL
        stderr = asyncio.subprocess.PIPE if stderr_pipe\
            else asyncio.subprocess.DEVNULL

        if self.is_running:
            _LOGGER.warning("FFmpeg is allready running!")
            return

        # set command line
        self._generate_ffmpeg_cmd(cmd, input_source, output, extra_cmd)

        # start ffmpeg
        _LOGGER.debug("Start FFmpeg with %s", str(self._argv))
        try:
            self._proc = yield from asyncio.create_subprocess_exec(
                *self._argv,
                loop=self._loop,
                stdin=asyncio.subprocess.PIPE,
                stdout=stdout,
                stderr=stderr
            )
        # pylint: disable=broad-except
        except Exception as err:
            _LOGGER.exception("FFmpeg fails %s", err)
            self._clear()
            return False

        return self._proc is not None

def run_task(context):
    """Run the task, sending stdout+stderr to files.

    https://github.com/python/asyncio/blob/master/examples/subprocess_shell.py

    Args:
        context (scriptworker.context.Context): the scriptworker context.

    Returns:
        int: exit code

    """
    loop = asyncio.get_event_loop()
    kwargs = {  # pragma: no branch
        'stdout': PIPE,
        'stderr': PIPE,
        'stdin': None,
        'close_fds': True,
        'preexec_fn': lambda: os.setsid(),
    }
    context.proc = await asyncio.create_subprocess_exec(*context.config['task_script'], **kwargs)
    loop.call_later(context.config['task_max_timeout'], max_timeout, context, context.proc, context.config['task_max_timeout'])

    tasks = []
    with get_log_filehandle(context) as log_filehandle:
        tasks.append(pipe_to_log(context.proc.stderr, filehandles=[log_filehandle]))
        tasks.append(pipe_to_log(context.proc.stdout, filehandles=[log_filehandle]))
        await asyncio.wait(tasks)
        exitcode = await context.proc.wait()
        status_line = "exit code: {}".format(exitcode)
        if exitcode == -11:
            status_line = "Automation Error: python exited with signal {}".format(exitcode)
        log.info(status_line)
        print(status_line, file=log_filehandle)

    context.proc = None
    return exitcode


# reclaim_task {{{1

def get_git_revision(path, ref="HEAD",
                           exec_function=asyncio.create_subprocess_exec):
    """Get the git revision of path.

    Args:
        path (str): the path to run ``git log -n1 --format=format:%H REF`` in.
        ref (str, optional): the ref to find the revision for.  Defaults to "HEAD"

    Returns:
        str: the revision found.

    Raises:
        ScriptWorkerRetryException: on failure.

    """
    proc = await exec_function(
        'git', "log", "-n1", "--format=format:%H", ref, cwd=path,
        stdout=PIPE, stderr=DEVNULL, stdin=DEVNULL, close_fds=True,
    )
    revision, err = await proc.communicate()
    exitcode = await proc.wait()
    if exitcode:
        raise ScriptWorkerRetryException(
            "Can't get repo revision at {}: {}!".format(path, err)
        )
    return revision.decode('utf-8').rstrip()

def ls(loop):
    proc = yield from asyncio.create_subprocess_exec("ls",
                                                     stdout=PIPE)
    while True:
        line = yield from proc.stdout.readline()
        if not line:
            break
        print("ls>>", line.decode('ascii').rstrip())
    try:
        proc.send_signal(signal.SIGINT)
    except ProcessLookupError:
        pass

def test_call(*args, timeout=None):
    proc = yield from asyncio.create_subprocess_exec(*args)
    try:
        exitcode = yield from asyncio.wait_for(proc.wait(), timeout)
        print("%s: exit code %s" % (' '.join(args), exitcode))
    except asyncio.TimeoutError:
        print("timeout! (%.1f sec)" % timeout)
        proc.kill()
        yield from proc.wait()

def communicate(cmdline, data=None, **kwargs):
    logging.debug('Running %s', ' '.join(cmdline))
    proc = await asyncio.create_subprocess_exec(
        *cmdline, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
        stderr=subprocess.PIPE, **kwargs)
    stdout, stderr = await proc.communicate(data)
    retcode = await proc.wait()
    return retcode, stdout, stderr