How to use the pexpect.replwrap function in pexpect

def test_async_replwrap_multiline(self):
        bash = replwrap.bash()
        coro = bash.run_command("echo '1 2\n3 4'", async_=True)
        res = run(coro)
        self.assertEqual(res.strip().splitlines(), ['1 2', '3 4'])

        # Should raise ValueError if input is incomplete
        coro = bash.run_command("echo '5 6", async_=True)
        try:
            run(coro)
        except ValueError:
            pass
        else:
            assert False, "Didn't raise ValueError for incomplete input"

        # Check that the REPL was reset (SIGINT) after the incomplete input
        coro = bash.run_command("echo '1 2\n3 4'", async_=True)
        res = run(coro)

def test_async_replwrap(self):
        bash = replwrap.bash()
        coro = bash.run_command("time", async_=True)
        res = run(coro)
        assert 'real' in res, res

def test_bash(self):
        bash = replwrap.bash()
        res = bash.run_command("time")
        assert 'real' in res, res

        # PAGER should be set to cat, otherwise man hangs
        res = bash.run_command('man sleep', timeout=5)
        assert 'SLEEP' in res, res

def test_python(self):
        if platform.python_implementation() == 'PyPy':
            raise unittest.SkipTest("This test fails on PyPy because of REPL differences")

        p = replwrap.python()
        res = p.run_command('4+7')
        assert res.strip() == '11'

        res = p.run_command('for a in range(3): print(a)\n')
        assert res.strip().splitlines() == ['0', '1', '2']

def get_shell(self):
        """Gets or creates the shell in which to run commands for the
        supplied demo
        """

        if self._shell is None:
            # Should we use spawn or spawnu?
            # The prompts used to be u'\[\]', but pylint prefers r'\[\]'.
            # Noting just in case this bites us later
            child = pexpect.spawnu('/bin/bash', env=self._env, echo=False, timeout=None)
            ps1 = PEXPECT_PROMPT[:5] + r'\[\]' + PEXPECT_PROMPT[5:]
            ps2 = PEXPECT_CONTINUATION_PROMPT[:5] + r'\[\]' + PEXPECT_CONTINUATION_PROMPT[5:]
            prompt_change = u"PS1='{0}' PS2='{1}' PROMPT_COMMAND=''".format(ps1, ps2)
            self._shell = replwrap.REPLWrapper(child, r'\$', prompt_change)
        return self._shell

def _start_java_repl(self):
        sig = signal.signal(signal.SIGINT, signal.SIG_DFL)
        try:
            self.javawrapper = replwrap.REPLWrapper(
                "{} -jar {}".format(
                    self._JAVA_COMMAND,
                    self._KULLA_LOCATION
                ),
                u'->',
                None
            )
        finally:
            signal.signal(signal.SIGINT, sig)

# "None" means we are executing code from a Jupyter cell by way of the run_command
            # in the do_execute() code below, so do incremental output.
            while True:
                pos = self.child.expect_exact([self.prompt, self.continuation_prompt, u'\r\n'],
                                              timeout=None)
                if pos == 2:
                    # End of line received
                    self.line_output_callback(self.child.before + '\n')
                else:
                    if len(self.child.before) != 0:
                        # prompt received, but partial line precedes it
                        self.line_output_callback(self.child.before)
                    break
        else:
            # Otherwise, use existing non-incremental code
            pos = replwrap.REPLWrapper._expect_prompt(self, timeout=timeout)

        # Prompt received, so return normally
        return pos

def _start_idl(self):
        # Signal handlers are inherited by forked processes, and we can't easily
        # reset it from the subprocess. Since kernelapp ignores SIGINT except in
        # message handlers, we need to temporarily reset the SIGINT handler here
        # so that IDL and its children are interruptible.
        sig = signal.signal(signal.SIGINT, signal.SIG_DFL)
        try:
            self._executable = find_executable("idl")
            self._child  = spawn(self._executable,timeout = 300)
            self.idlwrapper = replwrap.REPLWrapper(self._child,u"IDL> ",None)
        except:
            self._executable = find_executable("gdl")
            self._child  = spawn(self._executable,timeout = 300)
            self.idlwrapper = replwrap.REPLWrapper(self._child,u"GDL> ",None)
        finally:
            signal.signal(signal.SIGINT, sig)

        self.idlwrapper.run_command("!quiet=1 & defsysv,'!inline',0 & !more=0".rstrip(), timeout=None)
        # Compile IDL routines/functions
        dirname = os.path.dirname(os.path.abspath(__file__))
        self.idlwrapper.run_command(".compile "+dirname+"/snapshot.pro",timeout=None)

from subprocess import check_output
import os.path

import re
import signal

__version__ = '0.7.2'

version_pat = re.compile(r'version (\d+(\.\d+)+)')

from .images import (
    extract_image_filenames, display_data_for_image, image_setup_cmd
)

class IREPLWrapper(replwrap.REPLWrapper):
    """A subclass of REPLWrapper that gives incremental output
    specifically for bash_kernel.

    The parameters are the same as for REPLWrapper, except for one
    extra parameter:

    :param line_output_callback: a callback method to receive each batch
      of incremental output. It takes one string parameter.
    """
    def __init__(self, cmd_or_spawn, orig_prompt, prompt_change,
                 extra_init_cmd=None, line_output_callback=None):
        self.line_output_callback = line_output_callback
        replwrap.REPLWrapper.__init__(self, cmd_or_spawn, orig_prompt,
                                      prompt_change, extra_init_cmd=extra_init_cmd)

    def _expect_prompt(self, timeout=-1):

def ahead_source(org_config_filename, verbose=False):
    command = 'emacs'
    args = ['-batch',
            '-l', org_config_filename,
            '-l', 'ts-org-interaction.el',
            '--eval=(ts-repl)']
    spawn = pexpect.spawn(command, args, encoding='utf-8')
    emacs_repl_wrapper = pexpect.replwrap.REPLWrapper(
        spawn, "Lisp expression: ", None)

    source = o.Source.from_emacs_repl(emacs_repl_wrapper, verbose)
    source.get_tree = lambda: source.get_all_items(
        ['asana_id', 'asana_project_id'])

    return source