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 test_scp(event_loop):
    # ensure that asyncio.subprocess will work;
    try:
        asyncio.get_child_watcher().attach_loop(event_loop)
    except RuntimeError:
        pytest.skip('test_scp will always fail outside of MainThread')
    async with base.CleanModel() as model:
        await model.add_machine()
        await asyncio.wait_for(
            model.block_until(lambda: model.machines),
            timeout=240)
        machine = model.machines['0']
        await asyncio.wait_for(
            model.block_until(lambda: (machine.status == 'running' and
                                       machine.agent_status == 'started')),
            timeout=480)

        with NamedTemporaryFile() as f:
            f.write(b'testcontents')
            f.flush()
            await machine.scp_to(f.name, 'testfile', scp_opts='-p')

        with NamedTemporaryFile() as f:
            await machine.scp_from('testfile', f.name, scp_opts='-p')
            assert f.read() == b'testcontents' 

def test_ssh(event_loop):
    # ensure that asyncio.subprocess will work;
    try:
        asyncio.get_child_watcher().attach_loop(event_loop)
    except RuntimeError:
        pytest.skip('test_ssh will always fail outside of MainThread')
    async with base.CleanModel() as model:
        app = await model.deploy('ubuntu')

        await asyncio.wait_for(
            model.block_until(lambda: app.units),
            timeout=60)
        unit = app.units[0]
        await asyncio.wait_for(
            model.block_until(lambda: unit.machine),
            timeout=60)
        machine = unit.machine
        await asyncio.wait_for(
            model.block_until(lambda: (machine.status == 'running' and
                                       machine.agent_status == 'started')),
            timeout=480)
        output = await unit.ssh("echo test")
        assert(output == "test") 

def create(*args, **kwargs):
    """
    Runs a subprocess using asyncio.create_subprocess_exec and returns the
    process.
    """
    LOGGER.debug("proc.create: %r", args)
    proc = await asyncio.create_subprocess_exec(
        *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
        start_new_session=True,
        **kwargs,
    )
    proc.name = args[0]
    proc.args = args[1:]
    return proc 

def call_method(
            self, method, arg_signature, args, return_signature, returns
    ):
        cmd = 'busctl --user -- call '
        cmd += f'{service_name} {object_path} {object_interface} {method}'
        cmd += f' "{arg_signature}"'
        for i in args:
            cmd += f' {i}'

        create = asyncio.create_subprocess_shell(
                cmd, stdout=asyncio.subprocess.PIPE
        )

        proc = await create

        # Read one line of output
        data = await proc.stdout.readline()
        line = data.decode('ascii').rstrip()

        self.assertEqual(line, f'{return_signature} {returns}')

        await proc.wait() 

def __init__(
            self,
            pt_exec: Union[List[str], List[bytes]],
            state: Union[str, bytes, os.PathLike],
            *,
            exit_on_stdin_close: bool = True,
    ) -> None:
        """Create the adapter.

        Args:
            pt_exec: The pluggable transport command line to execute. This has
                to be a list of str / bytes, since
                :func:`asyncio.create_subprocess_exec` does not accept an
                entire command line as a string. On non-Windows platforms
                :func:`shlex.split` can be used to split a command line string
                into a list, while on Windows it's a bit more complicated.
            state: The state directory. This is a directory where the PT is
                allowed to store state. Either specify a path (which
                is not required to exist, in which case the PT will create
                the directory), or specify ``None`` to use a temporary
                directory created using :mod:`tempfile`.
            exit_on_stdin_close: Whether closing the PT's STDIN indicates the
                PT should gracefully exit.
        """
        if isinstance(pt_exec, (str, bytes)):
            self._pt_args = [pt_exec]
        else:
            self._pt_args = list(pt_exec)
        if state is not None:
            self._state = os.path.abspath(state)
        else:
            self._state = None
        self._exit_on_stdin_close = exit_on_stdin_close

        self._process: asyncio.subprocess.Process = None
        self._stdout_task: asyncio.Task = None
        self._ready = asyncio.Future()
        self._accepted_version: str = None
        self._transports: Dict[str, asyncio.Future] = {}
        self._stopping = False
        self._stack = contextlib.AsyncExitStack() 

def stop(self) -> None:
        """(async) Stop the PT executable.

        First try to signal a graceful exit by closing PT's STDIN (if
        enabled) and wait, then call
        :meth:`~asyncio.asyncio.subprocess.Process.terminate` and wait,
        then call
        :meth:`~asyncio.asyncio.subprocess.Process.kill`.
        """
        # Why does cross referencing asyncio.subprocess need
        # "asyncio.asyncio.subprocess"?
        self._check_running()
        self._stopping = True
        try:
            if self._exit_on_stdin_close:
                self._logger.debug('Closing PT stdin')
                self._process.stdin.close()
                try:
                    await asyncio.wait_for(
                        self._process.wait(), self._stdin_close_timeout)
                    self._logger.debug('PT exited after closing stdin')
                    return
                except asyncio.TimeoutError:
                    pass
            try:
                self._logger.debug('Terminating PT')
                self._process.terminate()
                try:
                    await asyncio.wait_for(
                        self._process.wait(), self._terminate_timeout)
                    self._logger.debug('PT exited after calling terminate()')
                    return
                except asyncio.TimeoutError:
                    pass
                self._logger.warning('Calling kill() on PT')
                self._process.kill()
                await self._process.wait()
            except ProcessLookupError:
                self._logger.info('PT process already exited')
        finally:
            await self._stack.aclose() 

def _kill_subprocess(self, proc: Optional[Process]) -> None:
        """Helper method; send SIGTERM/SIGKILL to a subprocess.

        This method first sends SIGTERM to the subprocess.  If the process hasn't terminated
        after a given timeout, it sends SIGKILL.

        Parameter
        ---------
        proc : Optional[Process]
            the process to attempt to terminate.  If None, this method does nothing.
        """
        if proc is not None:
            if proc.returncode is None:
                try:
                    proc.terminate()
                    try:
                        await asyncio.shield(asyncio.wait_for(proc.wait(), self._cancel_timeout))
                    except CancelledError:
                        pass

                    if proc.returncode is None:
                        proc.kill()
                        try:
                            await asyncio.shield(
                                asyncio.wait_for(proc.wait(), self._cancel_timeout))
                        except CancelledError:
                            pass
                except ProcessLookupError:
                    pass 

def batch_subprocess(self, proc_info_list):
        # type: (Sequence[ProcInfo]) -> Optional[Sequence[Union[int, Exception]]]
        """Run all given subprocesses in parallel.

        Parameters
        ----------
        proc_info_list : Sequence[ProcInfo]
            a list of process information.  Each element is a tuple of:

            args : Union[str, Sequence[str]]
                command to run, as string or list of string arguments.
            log : str
                log file name.
            env : Optional[Dict[str, str]]
                environment variable dictionary.  None to inherit from parent.
            cwd : Optional[str]
                working directory path.  None to inherit from parent.

        Returns
        -------
        results : Optional[Sequence[Union[int, Exception]]]
            if user cancelled the subprocesses, None is returned.  Otherwise, a list of
            subprocess return codes or exceptions are returned.
        """
        num_proc = len(proc_info_list)
        if num_proc == 0:
            return []

        coro_list = [self.async_new_subprocess(args, log, env, cwd) for args, log, env, cwd in
                     proc_info_list]

        return batch_async_task(coro_list) 

def batch_subprocess_flow(self, proc_info_list):
        # type: (Sequence[Sequence[FlowInfo]]) -> Optional[Sequence[Union[int, Exception]]]
        """Run all given subprocesses flow in parallel.

        Parameters
        ----------
        proc_info_list : Sequence[Sequence[FlowInfo]
            a list of process flow information.  Each element is a sequence of tuples of:

            args : Union[str, Sequence[str]]
                command to run, as string or list of string arguments.
            log : str
                log file name.
            env : Optional[Dict[str, str]]
                environment variable dictionary.  None to inherit from parent.
            cwd : Optional[str]
                working directory path.  None to inherit from parent.
            vfun : Sequence[Callable[[Optional[int], str], Any]]
                a function to validate if it is ok to execute the next process.  The output of the
                last function is returned.  The first argument is the return code, the second
                argument is the log file name.

        Returns
        -------
        results : Optional[Sequence[Any]]
            if user cancelled the subprocess flows, None is returned.  Otherwise, a list of
            flow return values or exceptions are returned.
        """
        num_proc = len(proc_info_list)
        if num_proc == 0:
            return []

        coro_list = [self.async_new_subprocess_flow(flow_info) for flow_info in proc_info_list]

        return batch_async_task(coro_list) 

def test_scp(event_loop):
    # ensure that asyncio.subprocess will work;
    try:
        asyncio.get_child_watcher().attach_loop(event_loop)
    except RuntimeError:
        pytest.skip('test_scp will always fail outside of MainThread')
    async with base.CleanModel() as model:
        app = await model.deploy('ubuntu')

        await asyncio.wait_for(
            model.block_until(lambda: app.units),
            timeout=60)
        unit = app.units[0]
        await asyncio.wait_for(
            model.block_until(lambda: unit.machine),
            timeout=60)
        machine = unit.machine
        await asyncio.wait_for(
            model.block_until(lambda: (machine.status == 'running' and
                                       machine.agent_status == 'started')),
            timeout=480)

        with NamedTemporaryFile() as f:
            f.write(b'testcontents')
            f.flush()
            await unit.scp_to(f.name, 'testfile')

        with NamedTemporaryFile() as f:
            await unit.scp_from('testfile', f.name)
            assert f.read() == b'testcontents' 

def exec_with_logging(*args):
    label = " ".join(args)
    LOGGER.debug("%s", label)

    proc = await asyncio.subprocess.create_subprocess_exec(
        *args, stdout=subprocess.PIPE, stderr=subprocess.PIPE
    )

    _, _, exit_code = await asyncio.gather(
        exec_with_logging_reader(label, proc.stdout),
        exec_with_logging_reader(label, proc.stderr),
        proc.wait(),
    )

    return exit_code 

def stop(proc):
    """
    Stops a subprocess
    """
    exit_code = await proc.wait()
    if exit_code != 0:
        raise RuntimeError(
            "'%s' exited with code %d: '%s'"
            % (
                getattr(proc, "name", "subprocess"),
                exit_code,
                getattr(proc, "data", "").rstrip(),
            )
        )
    return exit_code, proc 

def check_output(*args, **kwargs):
    """
    Runs a subprocess using asyncio.create_subprocess_exec and returns either
    its standard error or output.
    """
    proc = await create(*args, **kwargs)
    stdout, stderr = await get_output(proc)
    await stop(proc)
    return stdout or stderr 

def test_subprocess(event_loop):
    """Starting a subprocess should be possible."""
    proc = await asyncio.subprocess.create_subprocess_exec(
        sys.executable, '--version', stdout=asyncio.subprocess.PIPE)
    await proc.communicate() 

def test_subprocess_forbid(event_loop):
    """Starting a subprocess should be possible."""
    proc = await asyncio.subprocess.create_subprocess_exec(
        sys.executable, '--version', stdout=asyncio.subprocess.PIPE)
    await proc.communicate() 

def async_new_subprocess(self,
                                   args: Union[str, Sequence[str]],
                                   log: str,
                                   env: Optional[Dict[str, str]] = None,
                                   cwd: Optional[str] = None) -> Optional[int]:
        """A coroutine which starts a subprocess.

        If this coroutine is cancelled, it will shut down the subprocess gracefully using
        SIGTERM/SIGKILL, then raise CancelledError.

        Parameters
        ----------
        args : Union[str, Sequence[str]]
            command to run, as string or sequence of strings.
        log : str
            the log file name.
        env : Optional[Dict[str, str]]
            an optional dictionary of environment variables.  None to inherit from parent.
        cwd : Optional[str]
            the working directory.  None to inherit from parent.

        Returns
        -------
        retcode : Optional[int]
            the return code of the subprocess.
        """
        if isinstance(args, str):
            args = [args]

        # get log file name, make directory if necessary
        log = os.path.abspath(log)
        if os.path.isdir(log):
            raise ValueError('log file %s is a directory.' % log)
        os.makedirs(os.path.dirname(log), exist_ok=True)

        async with self._semaphore:
            proc = None
            with open(log, 'w') as logf:
                logf.write('command: %s\n' % (' '.join(args)))
                logf.flush()
                try:
                    proc = await asyncio.create_subprocess_exec(*args, stdout=logf,
                                                                stderr=subprocess.STDOUT,
                                                                env=env, cwd=cwd)
                    retcode = await proc.wait()
                    return retcode
                except CancelledError as err:
                    await self._kill_subprocess(proc)
                    raise err 

def cli(api_server, api_server_port, event_loop):
    @asyncio.coroutine
    def call_cli(*args, input=None, timeout=3.0):
        # Prepare environment
        env = os.environ.copy()
        env['THREEMA_TEST_API'] = str(api_server_port)
        test_api_mode = 'WARNING: Currently running in test mode!'

        # Call CLI in subprocess and get output
        parameters = [sys.executable, pytest.msgapi.cli_path] + list(args)
        if isinstance(input, str):
            input = input.encode('utf-8')

        # Create process
        create = asyncio.create_subprocess_exec(
            *parameters, env=env, stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT)
        process = yield from create

        # Wait for process to terminate
        coroutine = process.communicate(input=input)
        output, _ = yield from asyncio.wait_for(coroutine, timeout, loop=event_loop)

        # Process output
        output = output.decode('utf-8')
        if test_api_mode not in output:
            print(output)
            raise ValueError('Not running in test mode')

        # Strip leading empty lines and pydev debugger output
        rubbish = [
            'pydev debugger: process',
            'Traceback (most recent call last):',
            test_api_mode,
        ]
        lines = []
        skip_following_empty_lines = True
        for line in output.splitlines(keepends=True):
            if any((line.startswith(s) for s in rubbish)):
                skip_following_empty_lines = True
            elif not skip_following_empty_lines or len(line.strip()) > 0:
                lines.append(line)
                skip_following_empty_lines = False

        # Strip trailing empty lines
        empty_lines_count = 0
        for line in reversed(lines):
            if len(line.strip()) > 0:
                break
            empty_lines_count += 1
        if empty_lines_count > 0:
            lines = lines[:-empty_lines_count]
        output = ''.join(lines)

        # Check return code
        if process.returncode != 0:
            raise subprocess.CalledProcessError(process.returncode, parameters,
                                                output=output)
        return output
    return call_cli