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.'
python类ExceptionPexpect()的实例源码
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.')
depricated_test_filedescriptor.py 文件源码
项目：obsoleted-vpduserv
作者: InfraSIM
项目源码
文件源码
阅读 21
收藏 0
点赞 0
评论 0
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)
# --------------------------------------------------------------
# --------------------------------------------------------------
