def start(self) -> None:
        """(async) Start the PT executable and wait until it's ready.

        "Ready" means that all transports have finished initializing.
        """
        self._check_not_started()
        await self._pre_start()
        env = self._build_env()
        self._logger.debug('PT environment variables: %r', env)
        self._process = await asyncio.create_subprocess_exec(
            *self._pt_args,
            env=env,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=None,
        )
        self._logger.debug('Started PT subprocess: %r', self._process)
        self._stdout_task = asyncio.create_task(self._process_stdout())
        try:
            await self._ready
        except Exception:
            await self.stop()
            raise 

def _start_remote_processors_ssh(self, host, port, num_workers):
        # establish a reverse SSH tunnel from remote unix socket to
        # the local unix socket that our Unix server is listening on
        port_arg = ('-p', port) if port else ()
        remote_unix_path = util.get_temp_path()
        proc = await asyncio.create_subprocess_exec(
            'ssh',
            '-T', host,
            *port_arg,
            '-R', f'{remote_unix_path}:{self._unix_path}',
            stdin=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE)
        # spawn processors that will connect to the tunneled Unix server
        cmd = (
            f'distex_proc '
            f'-u {remote_unix_path} '
            f'-l {self._worker_loop} '
            f'-f {self._func_pickle} '
            f'-d {self._data_pickle} '
            f'& \n'.encode()) * num_workers
        proc.stdin.write(cmd)
        await proc.stdin.drain()
        self._ssh_tunnels.append(proc)
        self._total_workers += num_workers 

def build(versions_file, args, *, loop):

    with open(versions_file) as f:
        config = yaml.load(f.read())

    for action in args.actions:
        procs = []
        for version_map in config['versions']:
            args = shlex.split('make docker-{action} '
                               'IMAGE_NAME={image_name} '
                               'KAFKA_VERSION={kafka} '
                               'SCALA_VERSION={scala}'.format(
                                   action=action,
                                   image_name=config['image_name'],
                                   **version_map))
            proc = yield from asyncio.create_subprocess_exec(*args, loop=loop)
            procs.append(proc.wait())

        res = yield from asyncio.gather(*procs, loop=loop)
        if any(res):  # If any of statuses are not 0 return right away
            return res
    return res 

def _spawn(self):
        self._manager._stats_spawned += 1

        if self._proc is not None:
            self._manager._sup.create_task(self._kill_proc(self._proc))
            self._proc = None

        env = _ENV
        if debug.flags.server:
            env = {'EDGEDB_DEBUG_SERVER': '1', **_ENV}

        self._proc = await asyncio.create_subprocess_exec(
            *self._command_args,
            env=env,
            stdin=subprocess.DEVNULL)
        try:
            self._con = await asyncio.wait_for(
                self._server.get_by_pid(self._proc.pid),
                PROCESS_INITIAL_RESPONSE_TIMEOUT)
        except Exception:
            try:
                self._proc.kill()
            except ProcessLookupError:
                pass
            raise 

def can_sudo(password=None):
    if not password and app.sudo_pass:
        password = app.sudo_pass
    if password:
        opt = '-S'  # stdin
        password = '{}\n'.format(password).encode('utf8')
    else:
        opt = '-n'  # non-interactive
    proc = await asyncio.create_subprocess_exec('sudo', opt, '/bin/true',
                                                stdin=subprocess.PIPE,
                                                stdout=subprocess.DEVNULL,
                                                stderr=subprocess.DEVNULL)
    if password:
        await proc.communicate(password)
    else:
        await proc.wait()
    return proc.returncode == 0 

def run_openconnect(auth_info, host, args):
    command_line = [
        "sudo",
        "openconnect",
        "--cookie-on-stdin",
        "--servercert",
        auth_info.server_cert_hash,
        *args,
    ]

    logger.debug("Starting OpenConnect", command_line=command_line)
    proc = await asyncio.create_subprocess_exec(
        *command_line,
        host.vpn_url,
        stdin=asyncio.subprocess.PIPE,
        stdout=None,
        stderr=None,
    )
    proc.stdin.write(f"{auth_info.session_token}\n".encode())
    await proc.stdin.drain()
    await proc.wait() 

def start_daemon(*command, input=None, **kwargs):
        """Start a daemon for the VM

        This function take care to run it as appropriate user.

        :param command: command to run (array for
            :py:meth:`subprocess.check_call`)
        :param kwargs: args for :py:meth:`subprocess.check_call`
        :return: None
        """  # pylint: disable=redefined-builtin

        if os.getuid() == 0:
            # try to always have VM daemons running as normal user, otherwise
            # some files (like clipboard) may be created as root and cause
            # permission problems
            qubes_group = grp.getgrnam('qubes')
            command = ['runuser', '-u', qubes_group.gr_mem[0], '--'] + \
                      list(command)
        p = yield from asyncio.create_subprocess_exec(*command, **kwargs)
        stdout, stderr = yield from p.communicate(input=input)
        if p.returncode:
            raise subprocess.CalledProcessError(p.returncode, command,
                                                output=stdout, stderr=stderr) 

def test_120_start_standalone_with_cdrom_dom0(self):
        vmname = self.make_vm_name('appvm')
        self.vm = self.app.add_new_vm('StandaloneVM', label='red', name=vmname)
        self.loop.run_until_complete(self.vm.create_on_disk())
        self.vm.kernel = None
        self.vm.virt_mode = 'hvm'

        iso_path = self.create_bootable_iso()
        # start the VM using qvm-start tool, to test --cdrom option there
        p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
            'qvm-start', '--cdrom=dom0:' + iso_path, self.vm.name,
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT))
        (stdout, _) = self.loop.run_until_complete(p.communicate())
        self.assertEqual(p.returncode, 0, stdout)
        # check if VM do not crash instantly
        self.loop.run_until_complete(asyncio.sleep(5))
        self.assertTrue(self.vm.is_running())
        # Type 'poweroff'
        subprocess.check_call(['xdotool', 'search', '--name', self.vm.name,
                               'type', '--window', '%1', 'poweroff\r'])
        for _ in range(10):
            if not self.vm.is_running():
                break
            self.loop.run_until_complete(asyncio.sleep(1))
        self.assertFalse(self.vm.is_running()) 

def test_140_libvirt_events_reconnect(self):
        vmname = self.make_vm_name('vm')

        self.vm = self.app.add_new_vm(qubes.vm.appvm.AppVM,
            name=vmname, template=self.app.default_template,
            label='red')
        self.loop.run_until_complete(self.vm.create_on_disk())
        self.loop.run_until_complete(self.vm.start())
        p = self.loop.run_until_complete(asyncio.create_subprocess_exec(
            'systemctl', 'restart', 'libvirtd'))
        self.loop.run_until_complete(p.communicate())
        # check if events still works
        self.domain_paused_received = False
        self.vm.add_handler('domain-paused', self._test_140_on_domain_paused)
        self.loop.run_until_complete(self.vm.pause())
        self.loop.run_until_complete(self.vm.kill())
        self.loop.run_until_complete(asyncio.sleep(1))
        self.assertTrue(self.domain_paused_received,
            'event not received after libvirt restart') 

def init_cache_coro(log=logging.getLogger('qubes.storage.lvm')):
    cmd = _init_cache_cmd
    if os.getuid() != 0:
        cmd = ['sudo'] + cmd
    environ = os.environ.copy()
    environ['LC_ALL'] = 'C.utf8'
    p = yield from asyncio.create_subprocess_exec(*cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        close_fds=True, env=environ)
    out, err = yield from p.communicate()
    return_code = p.returncode
    if return_code == 0 and err:
        log.warning(err)
    elif return_code != 0:
        raise qubes.storage.StoragePoolException(err)

    return _parse_lvm_cache(out) 

def qubes_lvm_coro(cmd, log=logging.getLogger('qubes.storage.lvm')):
    ''' Call :program:`lvm` to execute an LVM operation

    Coroutine version of :py:func:`qubes_lvm`'''
    environ = os.environ.copy()
    environ['LC_ALL'] = 'C.utf8'
    if cmd[0] == "remove":
        pre_cmd = ['blkdiscard', '/dev/'+cmd[1]]
        p = yield from asyncio.create_subprocess_exec(*pre_cmd,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            close_fds=True, env=environ)
        _, _ = yield from p.communicate()
    cmd = _get_lvm_cmdline(cmd)
    p = yield from asyncio.create_subprocess_exec(*cmd,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        close_fds=True, env=environ)
    out, err = yield from p.communicate()
    return _process_lvm_output(p.returncode, out, err, log) 

def handle_request(self, reader, writer):
        req_host, req_port = writer.get_extra_info('peername')
        peername = f'{req_host}:{req_port}'
        self._logger.info(f'Connection from {peername}')
        data = await reader.readline()
        nw, port, worker_loop, func_pickle, data_pickle = data.split()
        num_workers = int(nw) or os.cpu_count()
        self._logger.info(
            f'Starting up {num_workers} processors for {peername}')

        # start processors that will connect back to the remote server
        asyncio.gather(
            *[asyncio.create_subprocess_exec(
                'distex_proc',
                '-H', req_host,
                '-p', port,
                '-l', worker_loop,
                '-f', func_pickle,
                '-d', data_pickle,
                stdout=None, stderr=None)
                for _ in range(num_workers)])

        writer.close() 

def test_user_code_execute():
    """
    User logs in, starts a server & executes code
    """
    # This *must* be localhost, not an IP
    # aiohttp throws away cookies if we are connecting to an IP!
    hub_url = 'http://localhost'
    username = secrets.token_hex(8)

    assert 0 == await (await asyncio.create_subprocess_exec(*TLJH_CONFIG_PATH, 'set', 'auth.type', 'dummyauthenticator.DummyAuthenticator')).wait()
    assert 0 == await (await asyncio.create_subprocess_exec(*TLJH_CONFIG_PATH, 'reload')).wait()

    async with User(username, hub_url, partial(login_dummy, password='')) as u:
            await u.login()
            await u.ensure_server()
            await u.start_kernel()
            await u.assert_code_output("5 * 4", "20", 5, 5)

            # Assert that the user exists
            assert pwd.getpwnam(f'jupyter-{username}') is not None 

def test_user_admin_add():
    """
    User is made an admin, logs in and we check if they are in admin group
    """
    # This *must* be localhost, not an IP
    # aiohttp throws away cookies if we are connecting to an IP!
    hub_url = 'http://localhost'
    username = secrets.token_hex(8)

    assert 0 == await (await asyncio.create_subprocess_exec(*TLJH_CONFIG_PATH, 'set', 'auth.type', 'dummyauthenticator.DummyAuthenticator')).wait()
    assert 0 == await (await asyncio.create_subprocess_exec(*TLJH_CONFIG_PATH, 'add-item', 'users.admin', username)).wait()
    assert 0 == await (await asyncio.create_subprocess_exec(*TLJH_CONFIG_PATH, 'reload')).wait()

    async with User(username, hub_url, partial(login_dummy, password='')) as u:
            await u.login()
            await u.ensure_server()

            # Assert that the user exists
            assert pwd.getpwnam(f'jupyter-{username}') is not None

            # Assert that the user has admin rights
            assert f'jupyter-{username}' in grp.getgrnam('jupyterhub-admins').gr_mem


# FIXME: Make this test pass 

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'
        ]
        cmd.extend(scp_opts.split() if isinstance(scp_opts, str) else scp_opts)
        cmd.extend([source, destination])
        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 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 _convert_fb_sticker(data: bytes, frames_per_row: int, frames_per_col: int
                                  ) -> Tuple[bytes, int, int]:
        ntf = NamedTemporaryFile
        with ntf(suffix=".png") as input_file, ntf(suffix=".gif") as output_file:
            input_file.write(data)
            with Image.open(input_file) as img:
                width, height = img.size
            width /= frames_per_row
            height /= frames_per_col
            proc = await asyncio.create_subprocess_exec(convert_cmd,
                                                        "-dispose", "Background",
                                                        input_file.name,
                                                        "-crop", f"{width}x{height}",
                                                        "+adjoin", "+repage", "-adjoin",
                                                        "-loop", "0",
                                                        output_file.name)
            await proc.wait()
            return output_file.read(), width, height 

def safety_check(package: str, version: str) -> int:
    pinned = f"{package}=={version}"

    proc = await asyncio.create_subprocess_exec(
        sys.executable,
        "-m",
        "safety",
        "check",
        "--stdin",
        "--json",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    stdout, _stderr = await proc.communicate(pinned.encode() + b"\n")

    issues = json.loads(stdout)

    return len(issues) 

def run_npm_audit(pkg: str) -> Dict[str, Any]:
    """
    CLI usage: dffml service dev run -log debug shouldi.npm_audit:run_npm_audit -pkg .
    """
    proc = await asyncio.create_subprocess_exec(
        "npm",
        "audit",
        "--json",
        cwd=pkg,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )

    stdout, stderr = await proc.communicate()
    if proc.returncode != 0 and stderr:
        raise NPMAuditError(stderr.decode())

    npm_audit_op = stdout.decode()
    npm_audit_op = json.loads(npm_audit_op)
    result = npm_audit_op["metadata"]["vulnerabilities"]
    return {"report": result} 

def git_clone(git_url, path, verbose=False):
    """Clone git repository at $git_url to $path. Return True if succesful,
    otherwise False."""
    if verbose:
        print("Cloning {}...".format(git_url))
    if os.path.exists(os.path.join(path, ".git")):
        # get rid of local repo if it already exists
        shutil.rmtree(path)

    os.makedirs(path, exist_ok=True)

    proc_env = os.environ.copy()
    proc_env["GIT_TERMINAL_PROMPT"] = "0"
    git_proc = await asyncio.create_subprocess_exec(
        "git", "clone", git_url, path, stderr=asyncio.subprocess.PIPE, env=proc_env
    )
    stdout, stderr = await git_proc.communicate()

    if git_proc.returncode != 0:
        # remove created directory if it's empty
        try:
            os.rmdir(path)
        except OSError:
            pass

        verb_msg("{}:\n{}".format(git_url, stderr.decode("utf-8")))
        return False
    elif verbose:
        print("Cloned {}".format(git_url))
    return True 

def __init__(self, *args, popen=asyncio.create_subprocess_exec, **kwargs):
        super(AioProcessProvider, self).__init__(*args, **kwargs, popen=popen) 

def process_provider():
    def _f(profile_name='default', loaded_config=None, invoked_process=None):
        load_config = mock.Mock(return_value=loaded_config)
        popen_mock = mock.Mock(return_value=invoked_process or mock.Mock(),
                               spec=asyncio.create_subprocess_exec)
        return popen_mock, credentials.AioProcessProvider(profile_name,
                                                          load_config,
                                                          popen=popen_mock)
    return _f 

def run_app(bootnode, listen_host, listen_port, max_peers, privkey):
    cmd = (
        "python trinity_discovery.py "
        "--bootnode={} "
        "--listen_host={} "
        "--listen_port={} "
        "--max_peers={} "
        "--privkey={}".format(bootnode, listen_host, listen_port, max_peers, privkey)
    )
    return await asyncio.create_subprocess_exec(
        *cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT
    ) 

def run_node(bootnode, privkey, listen_port, max_peers, logging_level):
    cmd = (
        "python paragon_node.py "
        "--bootnode={} "
        "--privkey={} "
        "--listen_port={} "
        "--max_peers={} "
        "--logging_level={}".format(
            bootnode, privkey, listen_port, max_peers, logging_level
        )
    )
    return await asyncio.create_subprocess_exec(
        *cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT
    ) 

def run_master(config_file, extra_cmd):
    cmd = "{} -u master.py --cluster_config={}".format(PYTHON, config_file)
    cmd += extra_cmd
    return await asyncio.create_subprocess_exec(
        *cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT
    ) 

def run_slave(config_file, id, profile):
    cmd = "{} -u slave.py --cluster_config={} --node_id={}".format(
        PYTHON, config_file, id
    )
    if profile:
        cmd += " --enable_profiler=true"
    return await asyncio.create_subprocess_exec(
        *cmd.split(" "), stdout=subprocess.PIPE, stderr=subprocess.STDOUT
    )