Python pexpect 模块，ExceptionPexpect() 实例源码

def _create_named_pipe(template, sids=None):
    """INTERNAL: create a named pipe."""
    if sids is None:
        sattrs = None
    else:
        sattrs = _create_security_attributes(*sids)
    for i in range(100):
        name = template % random.randint(0, 999999)
        try:
            pipe = CreateNamedPipe(name, PIPE_ACCESS_DUPLEX,
                                   0, 1, 1, 1, 100000, sattrs)
            SetHandleInformation(pipe, HANDLE_FLAG_INHERIT, 0)
        except WindowsError, e:
            if e.winerror != ERROR_PIPE_BUSY:
                raise
        else:
            return pipe, name
    raise ExceptionPexpect, 'Could not create pipe after 100 attempts.'

def OFF_test_run_out_of_pty (self):
        '''This assumes that the tested platform has < 10000 pty devices.
        This test currently does not work under Solaris.
        Under Solaris it runs out of file descriptors first and
        ld.so starts to barf:
            ld.so.1: pt_chmod: fatal: /usr/lib/libc.so.1: Too many open files
        '''
        plist=[]
        for count in range (0,10000):
                try:
                        plist.append (pexpect.spawn('ls -l'))
                except pexpect.ExceptionPexpect:
                        for c in range (0, count):
                            plist[c].close()
                        return
                except Exception:
                        err = sys.exc_info()[1]
                        self.fail ('Expected ExceptionPexpect. ' + str(err))
        self.fail ('Could not run out of pty devices. This may be OK.')

def run(self):
        try:
            self.term = console.Terminal(
                hops.list_hops(),
                target_host = self.host,
                meta = self.pid,
                tag = self.tag,
                prompt_re = r'\(gdb\)\ \r\n',
                exit_re = r'&"quit\n"|\^exit',
            )
            self.term.connect()

            gdb_response = self.term.query(self.cmd)
            try:
                self.mi_response = mi_interface.parse(gdb_response)
            except pexpect.ExceptionPexpect as e:
                raise errors.CommandFailedError('attach', 'attach', e)
        except Exception as e:
            self.error = e

def OFF_test_run_out_of_pty (self):
        '''This assumes that the tested platform has < 10000 pty devices.
        This test currently does not work under Solaris.
        Under Solaris it runs out of file descriptors first and
        ld.so starts to barf:
            ld.so.1: pt_chmod: fatal: /usr/lib/libc.so.1: Too many open files
        '''
        plist=[]
        for count in range (0,10000):
                try:
                        plist.append (pexpect.spawn('ls -l'))
                except pexpect.ExceptionPexpect:
                        for c in range (0, count):
                            plist[c].close()
                        return
                except Exception:
                        err = sys.exc_info()[1]
                        self.fail ('Expected ExceptionPexpect. ' + str(err))
        self.fail ('Could not run out of pty devices. This may be OK.')

def test_fd_isatty (self):
        fd = os.open ('TESTDATA.txt', os.O_RDONLY)
        s = pexpect.spawn (fd)
        assert not s.isatty()
        os.close(fd)

###    def test_close_does_not_close_fd (self):
###        '''Calling close() on a pexpect.spawn object should not
###                close the underlying file descriptor.
###        '''
###        fd = os.open ('TESTDATA.txt', os.O_RDONLY)
###        s = pexpect.spawn (fd)
###        try:
###            s.close()
###            self.fail('Expected an Exception.')
###        except pexpect.ExceptionPexpect, e:
###            pass

def test_fd_isatty (self):
        fd = os.open ('TESTDATA.txt', os.O_RDONLY)
        s = pexpect.spawn (fd)
        assert not s.isatty()
        os.close(fd)

###    def test_close_does_not_close_fd (self):
###        '''Calling close() on a pexpect.spawn object should not
###                close the underlying file descriptor.
###        '''
###        fd = os.open ('TESTDATA.txt', os.O_RDONLY)
###        s = pexpect.spawn (fd)
###        try:
###            s.close()
###            self.fail('Expected an Exception.')
###        except pexpect.ExceptionPexpect, e:
###            pass

def _start(self, startup_args=None):
        if not startup_args:
            startup_args = self._create_startup_arg_list(self._current_color,
                **self.init_kwargs)
        try:
            previous_instances = pexpect.run('pgrep -d, -u %s xflux' % pexpect.run('whoami')).strip()
            if previous_instances != "":
                for process in previous_instances.split(","):
                    pexpect.run('kill -9 %s' % process)

            self._xflux = pexpect.spawn("xflux", startup_args)
                    #logfile=file("tmp/xfluxout.txt",'w'))

        except pexpect.ExceptionPexpect:
            raise FileNotFoundError(
                    "\nError: Please install xflux in the PATH \n")

def quit(self):

        # Return immediately of already dead
        if not hasattr(self, '_process') or not self._process.isalive():
            # self._log.debug('ModelSim has already been stopped')
            # try:
            #   # I am being pedantic here, in case, for any reason, it wasn't done yet
            #   self.__instances.remove(self)
            # except KeyError:
            #   pass
            return

        try:
            self.execute('quit')
        except pexpect.ExceptionPexpect:
            pass

        # Just in case
        self._process.terminate(True)

        self.__instances.remove(self)
    # --------------------------------------------------------------

    # --------------------------------------------------------------

def connect(self, cmd, timeout=30):
        """
        Returns:
            pexpect.spwan object: on success.

            False: on Failure

        Args:
            cmd (string): command to spawn.

            timeout (int): default 30 seconds.

        Examples:
            EX1::

                expect = ExpectShell()
                expect.connect('telnet shark 2305', timeout=60)

            Ex2::

                expect = ExpectShell()
                expect.connect('gdb a.out')
        """
        if not cmd:
            return False
        if self.shell is None:
            try:
                log.command("%s" % cmd)
                self.shell = pexpect.spawn(cmd, timeout=timeout,
                                           echo=False)
                return self.shell
            except pexpect.ExceptionPexpect:
                log.error("Failed to spwan %s " % cmd)
                return False

def kill(self, signo):
        """Send a signal to the child (not available on Windows)."""
        raise ExceptionPexpect, 'Signals are not availalbe on Windows'

def __init__ (self, fd, args=[], timeout=30, maxread=2000, searchwindowsize=None, logfile=None):

        '''This takes a file descriptor (an int) or an object that support the
        fileno() method (returning an int). All Python file-like objects
        support fileno(). '''

        ### TODO: Add better handling of trying to use fdspawn in place of spawn
        ### TODO: (overload to allow fdspawn to also handle commands as spawn does.

        if type(fd) != type(0) and hasattr(fd, 'fileno'):
            fd = fd.fileno()

        if type(fd) != type(0):
            raise ExceptionPexpect('The fd argument is not an int. If this is a command string then maybe you want to use pexpect.spawn.')

        try: # make sure fd is a valid file descriptor
            os.fstat(fd)
        except OSError:
            raise ExceptionPexpect('The fd argument is not a valid file descriptor.')

        self.args = None
        self.command = None
        spawn.__init__(self, None, args, timeout, maxread, searchwindowsize, logfile)
        self.child_fd = fd
        self.own_fd = False
        self.closed = False
        self.name = '<file descriptor %d>' % fd

def terminate (self, force=False):  # pragma: no cover
        raise ExceptionPexpect('This method is not valid for file descriptors.')

def test_picking(self):
        e = ExceptionPexpect('Oh noes!')
        clone = pickle.loads(pickle.dumps(e))
        self.assertEqual(e.value, clone.value)

def test_bad_child_pid(self):
        child = pexpect.spawn('cat')
        child.terminate(force=1)
        child.terminated = 0 # Force invalid state to test code
        try:
            child.isalive()
        except pexpect.ExceptionPexpect:
            pass
        else:
            self.fail ("child.isalive() should have raised a pexpect.ExceptionPexpect")
        child.terminated = 1 # Force back to valid state so __del__ won't complain

def test_bad_arguments (self):
        '''This tests that we get a graceful error when passing bad arguments.'''
        with self.assertRaises(pexpect.ExceptionPexpect):
            pexpect.spawn(1)

        with self.assertRaises(TypeError):
            # should use pexpect.spawn('ls', ['-la'])
            pexpect.spawn('ls', '-la')

        with self.assertRaises(ValueError):
            p = pexpect.spawn('cat', timeout=5)
            p.close()
            p.read_nonblocking(size=1, timeout=3)

def test_expect_wait (self):
        '''This tests that calling wait on a finished process works as expected.
        '''
        p = pexpect.spawn('sleep 3')
        assert p.isalive()
        p.wait()
        assert not p.isalive()

        p = pexpect.spawn('sleep 3')
        assert p.isalive()
        p.kill(9)
        time.sleep(1)
        with self.assertRaises(pexpect.ExceptionPexpect):
            p.wait()

def __init__ (self, fd, args=[], timeout=30, maxread=2000, searchwindowsize=None, logfile=None):

        '''This takes a file descriptor (an int) or an object that support the
        fileno() method (returning an int). All Python file-like objects
        support fileno(). '''

        ### TODO: Add better handling of trying to use fdspawn in place of spawn
        ### TODO: (overload to allow fdspawn to also handle commands as spawn does.

        if type(fd) != type(0) and hasattr(fd, 'fileno'):
            fd = fd.fileno()

        if type(fd) != type(0):
            raise ExceptionPexpect('The fd argument is not an int. If this is a command string then maybe you want to use pexpect.spawn.')

        try: # make sure fd is a valid file descriptor
            os.fstat(fd)
        except OSError:
            raise ExceptionPexpect('The fd argument is not a valid file descriptor.')

        self.args = None
        self.command = None
        spawn.__init__(self, None, args, timeout, maxread, searchwindowsize, logfile)
        self.child_fd = fd
        self.own_fd = False
        self.closed = False
        self.name = '<file descriptor %d>' % fd

def terminate (self, force=False):  # pragma: no cover
        raise ExceptionPexpect('This method is not valid for file descriptors.')

def __init__ (self, fd, args=[], timeout=30, maxread=2000, searchwindowsize=None, logfile=None):

        '''This takes a file descriptor (an int) or an object that support the
        fileno() method (returning an int). All Python file-like objects
        support fileno(). '''

        ### TODO: Add better handling of trying to use fdspawn in place of spawn
        ### TODO: (overload to allow fdspawn to also handle commands as spawn does.

        if type(fd) != type(0) and hasattr(fd, 'fileno'):
            fd = fd.fileno()

        if type(fd) != type(0):
            raise ExceptionPexpect('The fd argument is not an int. If this is a command string then maybe you want to use pexpect.spawn.')

        try: # make sure fd is a valid file descriptor
            os.fstat(fd)
        except OSError:
            raise ExceptionPexpect('The fd argument is not a valid file descriptor.')

        self.args = None
        self.command = None
        spawn.__init__(self, None, args, timeout, maxread, searchwindowsize, logfile)
        self.child_fd = fd
        self.own_fd = False
        self.closed = False
        self.name = '<file descriptor %d>' % fd

def terminate (self, force=False):  # pragma: no cover
        raise ExceptionPexpect('This method is not valid for file descriptors.')

def test_picking(self):
        e = ExceptionPexpect('Oh noes!')
        clone = pickle.loads(pickle.dumps(e))
        self.assertEqual(e.value, clone.value)

def test_bad_child_pid(self):
        child = pexpect.spawn('cat')
        child.terminate(force=1)
        child.terminated = 0 # Force invalid state to test code
        try:
            child.isalive()
        except pexpect.ExceptionPexpect:
            pass
        else:
            self.fail ("child.isalive() should have raised a pexpect.ExceptionPexpect")
        child.terminated = 1 # Force back to valid state so __del__ won't complain

def test_bad_arguments (self):
        '''This tests that we get a graceful error when passing bad arguments.'''
        with self.assertRaises(pexpect.ExceptionPexpect):
            pexpect.spawn(1)

        with self.assertRaises(TypeError):
            # should use pexpect.spawn('ls', ['-la'])
            pexpect.spawn('ls', '-la')

        with self.assertRaises(ValueError):
            p = pexpect.spawn('cat', timeout=5)
            p.close()
            p.read_nonblocking(size=1, timeout=3)

def test_expect_wait (self):
        '''This tests that calling wait on a finished process works as expected.
        '''
        p = pexpect.spawn('sleep 3')
        assert p.isalive()
        p.wait()
        assert not p.isalive()

        p = pexpect.spawn('sleep 3')
        assert p.isalive()
        p.kill(9)
        time.sleep(1)
        with self.assertRaises(pexpect.ExceptionPexpect):
            p.wait()

def quit(self):

        # Return immediately of already dead
        if not hasattr(self, '_process') or not self._process.isalive():
            self._log.debug('Vivado has already been stopped')
            # try:
            #   # I am being pedantic here, in case, for any reason, it wasn't done yet
            #   self.__instances.remove(self)
            # except KeyError:
            #   pass
            return

        self._log.debug('Shutting Vivado down')
        try:
            self.execute('quit')
        except pexpect.ExceptionPexpect as e:
            pass

        # Just in case
        self._process.terminate(True)

        # Remove self from the list of instances
        self.__instances.remove(self)
    # --------------------------------------------------------------

    # --------------------------------------------------------------