How to use the pexpect.popen_spawn.PopenSpawn function in pexpect

def test_expect_timeout(self):
        p = PopenSpawn('cat', timeout=5)
        p.expect(pexpect.TIMEOUT)  # This tells it to wait for timeout.
        self.assertEqual(p.after, pexpect.TIMEOUT)

def test_unexpected_eof(self):
        p = PopenSpawn('ls -l /bin')
        try:
            p.expect('_Z_XY_XZ')  # Probably never see this in ls output.
        except pexpect.EOF:
            pass
        else:
            self.fail('Expected an EOF exception.')

args = ['../run.py'] + args
    options, unused_args = parse_cmdline()
    if options.coverage:
        args = ['./coverage.py', '-x', '-p'] + args
    if options.log:
        logfile = open(options.log, 'a', 0o644)
        args += ['--debug']
        print('Launching:', str(args), file=logfile)
    else:
        logfile = None

    if input_data is None:
        child = pexpect.spawn(args[0], args=args[1:],
                              encoding='utf-8', logfile=logfile)
    else:
        child = PopenSpawn(args, encoding='utf-8', logfile=logfile)
        child.send(input_data)
        child.sendeof()
    return child

def test_expect_eof(self):
        the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'],
                                       stdout=subprocess.PIPE).communicate()[0].rstrip()
        p = PopenSpawn('ls -l /bin')
        # This basically tells it to read everything. Same as pexpect.run()
        # function.
        p.expect(pexpect.EOF)
        the_new_way = p.before.rstrip()
        assert the_old_way == the_new_way, len(the_old_way) - len(the_new_way)

def test_expect_exact(self):
        the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'],
                                       stdout=subprocess.PIPE).communicate()[0].rstrip()
        p = PopenSpawn('ls -l /bin')
        the_new_way = b''
        while 1:
            i = p.expect_exact([b'\n', pexpect.EOF])
            the_new_way = the_new_way + p.before
            if i == 1:
                break
            the_new_way += b'\n'
        the_new_way = the_new_way.rstrip()

        assert the_old_way == the_new_way, len(the_old_way) - len(the_new_way)
        p = PopenSpawn('echo hello.?world')
        i = p.expect_exact(b'.?')
        self.assertEqual(p.before, b'hello')
        self.assertEqual(p.after, b'.?')

def test_expect_exact(self):
        the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'],
                                       stdout=subprocess.PIPE).communicate()[0].rstrip()
        p = PopenSpawn('ls -l /bin')
        the_new_way = b''
        while 1:
            i = p.expect_exact([b'\n', pexpect.EOF])
            the_new_way = the_new_way + p.before
            if i == 1:
                break
            the_new_way += b'\n'
        the_new_way = the_new_way.rstrip()

        assert the_old_way == the_new_way, len(the_old_way) - len(the_new_way)
        p = PopenSpawn('echo hello.?world')
        i = p.expect_exact(b'.?')
        self.assertEqual(p.before, b'hello')
        self.assertEqual(p.after, b'.?')

def run(self, block=True):
        """Runs the given command, with or without pexpect functionality enabled."""
        self.blocking = block

        # Use subprocess.
        if self.blocking:
            s = subprocess.Popen(self._popen_args, **self._default_popen_kwargs)

        # Otherwise, use pexpect.
        else:
            s = PopenSpawn(self._popen_args, **self._default_pexpect_kwargs)
        self.subprocess = s
        self.was_run = True

try:
            db.close_old_connections()
        except django.core.exceptions.ImproperlyConfigured:
            # The Django environment is not correctly setup. This might be
            # because we are calling Django management commands with subprocess
            # calls. In this case the subprocess we are calling will handle
            # closing of old connections.
            pass
        instance_connection_string = '{0}:{1}:{2}'.format(
            project_id, region, instance_name)
        instance_flag = '-instances={}=tcp:{}'.format(
            instance_connection_string, port)
        if cloud_sql_proxy_path is None:
            cloud_sql_proxy_path = shutil.which('cloud_sql_proxy')
            assert cloud_sql_proxy_path, 'could not find cloud_sql_proxy_path'
        process = popen_spawn.PopenSpawn([cloud_sql_proxy_path, instance_flag])
        try:
            # Make sure cloud sql proxy is started before doing the real work
            process.expect('Ready for new connections', timeout=60)
            yield
        except pexpect.exceptions.TIMEOUT:
            raise DatabaseError(
                ('Cloud SQL Proxy was unable to start after 60 seconds. Output '
                 'of cloud_sql_proxy: \n{}').format(process.before))
        except pexpect.exceptions.EOF:
            raise DatabaseError(
                ('Cloud SQL Proxy exited unexpectedly. Output of '
                 'cloud_sql_proxy: \n{}').format(process.before))
        finally:
            process.kill(signal.SIGTERM)

def __init__(self, engine=''):
        if not engine:
            engine = './parser'
        self.p = PopenSpawn(engine, timeout=TIME_OUT_SECOND, encoding="utf-8")
        self.pgn = ''
        self.db = ''

popen_kwargs['cwd'] = cwd
            if env:
                popen_kwargs['env'].update(env)
            s = subprocess.Popen(self._popen_args, **popen_kwargs)
        # Otherwise, use pexpect.
        else:
            pexpect_kwargs = self._default_pexpect_kwargs.copy()
            if binary:
                pexpect_kwargs['encoding'] = None
            if cwd:
                pexpect_kwargs['cwd'] = cwd
            if env:
                pexpect_kwargs['env'].update(env)
            # Enable Python subprocesses to work with expect functionality.
            pexpect_kwargs['env']['PYTHONUNBUFFERED'] = '1'
            s = PopenSpawn(self._popen_args, **pexpect_kwargs)
        self.subprocess = s
        self.was_run = True