def main ():
while True:
ps = pexpect.spawn ('ps')
time.sleep (1)
index = ps.expect (['/usr/bin/ssh', pexpect.EOF, pexpect.TIMEOUT])
if index == 2:
print('TIMEOUT in ps command...')
print(str(ps))
time.sleep (13)
if index == 1:
print(time.asctime(), end=' ')
print('restarting tunnel')
start_tunnel ()
time.sleep (11)
print('tunnel OK')
else:
# print 'tunnel OK'
time.sleep (7)
python类spawn()的实例源码
def test_expect (self):
the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'],
stdout=subprocess.PIPE).communicate()[0].rstrip()
p = pexpect.spawn('ls -l /bin')
the_new_way = b''
while 1:
i = p.expect ([b'\n', pexpect.EOF])
the_new_way = the_new_way + p.before
if i == 1:
break
the_new_way = the_new_way.rstrip()
the_new_way = the_new_way.replace(b'\r\n', b'\n'
).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip()
the_old_way = the_old_way.replace(b'\r\n', b'\n'
).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip()
assert the_old_way == the_new_way, hex_diff(the_old_way, the_new_way)
def test_expect_exact (self):
the_old_way = subprocess.Popen(args=['ls', '-l', '/bin'],
stdout=subprocess.PIPE).communicate()[0].rstrip()
p = pexpect.spawn('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 = the_new_way.replace(b'\r\n', b'\n'
).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip()
the_old_way = the_old_way.replace(b'\r\n', b'\n'
).replace(b'\r', b'\n').replace(b'\n\n', b'\n').rstrip()
assert the_old_way == the_new_way, hex_diff(the_old_way, the_new_way)
p = pexpect.spawn('echo hello.?world')
i = p.expect_exact(b'.?')
self.assertEqual(p.before, b'hello')
self.assertEqual(p.after, b'.?')
def test_signal_handling(self):
'''
This tests the error handling of a signal interrupt (usually a
SIGWINCH generated when a window is resized), but in this test, we
are substituting an ALARM signal as this is much easier for testing
and is treated the same as a SIGWINCH.
To ensure that the alarm fires during the expect call, we are
setting the signal to alarm after 1 second while the spawned process
sleeps for 2 seconds prior to sending the expected output.
'''
def noop(x, y):
pass
signal.signal(signal.SIGALRM, noop)
p1 = pexpect.spawn('%s sleep_for.py 2' % self.PYTHONBIN, timeout=5)
p1.expect('READY')
signal.alarm(1)
p1.expect('END')
def get_proc(shell, homedir):
return pexpect.spawn(
shell[0], list(shell[1:]),
timeout=5,
env={
'COVERAGE_PROCESS_START': os.environ.get(
'COVERAGE_PROCESS_START', '',
),
'PS1': PS1,
'TOP': os.environ.get('TOP', ''),
'HOME': str(homedir),
'PATH': os.path.dirname(sys.executable) + os.defpath
},
)
def spawn(command, args):
print command, ' '.join(args[1:])
ret = os.spawnvp(os.P_WAIT, command, args)
if ret != 0:
raise RuntimeError("could not run " + command)
# Subclass pexpect.spawn to add logging of expect() calls
def expect(self, pattern, *args, **kwargs):
print >>self.structured_log_f, "expect(" + repr(pattern) + ")"
r = pexpect.spawn.expect(self, pattern, *args, **kwargs)
print >>self.structured_log_f, "match(" + repr(self.match.group(0)) + ")"
return r
# Subclass urllib.FancyURLopener so that we can catch
# HTTP 404 errors
def make_iso(self):
self.download()
spawn(makefs[0], makefs + \
[self.iso_path(), os.path.dirname(os.path.realpath(os.path.join(self.download_local_mi_dir(), self.arch())))])
self.tempfiles.append(self.iso_path())
# Get the architecture name. This is a hardcoded default for use
# by the obsolete subclasses; the "URL" class overrides it.
def __del__(self):
print "destroying domU", self.name
spawn(self.frontend, [self.frontend, "destroy", self.name])
def slog(self, message):
slog_info(self.structured_log_f, message)
# Wrapper around pexpect.spawn to let us log the command for
# debugging. Note that unlike os.spawnvp, args[0] is not
# the name of the command.
def testStartupOptions(self):
cmd = "python {0} ".format(resource_file("ravel.py"))
p = pexpect.spawn(cmd + "--help")
p.expect("Usage")
p.sendeof()
time.sleep(1)
p = pexpect.spawn(cmd + "--topo=single,3")
p.expect("ravel>")
p.sendeof()
time.sleep(1)
p = pexpect.spawn(cmd + "--topo=single,3 --onlydb")
p.expect("ravel>")
p.sendline("m")
p.sendline("net")
p.expect("no CLI available")
p.sendline("exit")
p.sendeof()
time.sleep(1)
p = pexpect.spawn(cmd + "--topo single,3 --noctl")
p.expect("Unable to contact the remote controller")
p.expect("ravel>")
p.sendline("exit")
p.sendeof()
def testCommands(self):
p = pexpect.spawn(self.ravelCmd)
p.expect("ravel>")
p.sendline("exit")
p.expect(pexpect.EOF)
def testStartup(self):
cmd = "python {0} ".format(resource_file("ravel.py"))
p = pexpect.spawn(cmd + "--help")
p.expect("Usage")
p.sendeof()
time.sleep(1)
p = pexpect.spawn(cmd + "--topo=single,2")
p.expect("ravel>")
p.sendline("exit")
p.sendeof()
time.sleep(1)
p = pexpect.spawn(cmd + "--topo=single,2 --onlydb")
p.expect("ravel>")
p.sendline("exit")
p.sendeof()
time.sleep(1)
p = pexpect.spawn(cmd + "--topo=single,2 --verbosity=debug")
p.expect("DEBUG")
p.sendline("exit")
p.sendeof()
time.sleep(1)
# Part 2 - Ravel Commands
def kernel_conda(kernel):
"""Run which conda in the test fixture kernel using Jupyter console."""
jupyter = pexpect.spawn('jupyter', [
'console',
'--kernel', kernel.name]
)
jupyter.expect('In.*:')
jupyter.sendline('!which conda')
# input echo
jupyter.readline()
# path output
path = jupyter.readline()
jupyter.close()
return path.decode('utf-8')
def brute(word):
print "Trying:",word
child = pexpect.spawn ('su')
child.expect ('Password: ')
child.sendline (word)
i = child.expect (['.+\s#\s',LOGIN_ERROR])
if i == 0:
print "\n\t[!] Root Password:",word
child.sendline ('whoami')
print child.before
child.interact()
#if i == 1:
#print "Incorrect Password"
def brute(word):
print "Trying:",word
child = pexpect.spawn ('su '+user)
child.expect ('Password: ')
child.sendline (word)
i = child.expect([LOGIN_ERROR, pexpect.TIMEOUT], timeout=5)
if i == 1:
print "\n\t[!] Password:",word
child.sendline ('whoami')
print child.before
child.interact()
#if i = 0:
#print "Incorrect Password"
def _try_passwordless_openssh(server, keyfile):
"""Try passwordless login with shell ssh command."""
if pexpect is None:
raise ImportError("pexpect unavailable, use paramiko")
cmd = 'ssh -f '+ server
if keyfile:
cmd += ' -i ' + keyfile
cmd += ' exit'
# pop SSH_ASKPASS from env
env = os.environ.copy()
env.pop('SSH_ASKPASS', None)
ssh_newkey = 'Are you sure you want to continue connecting'
p = pexpect.spawn(cmd, env=env)
while True:
try:
i = p.expect([ssh_newkey, _password_pat], timeout=.1)
if i==0:
raise SSHException('The authenticity of the host can\'t be established.')
except pexpect.TIMEOUT:
continue
except pexpect.EOF:
return True
else:
return False
def setUp(self):
self.rsf = pexpect.spawn('python {}'.format(self.cli_path))
self.rsf.send('\r\n')
self.rsf.expect_exact(self.raw_prompt, timeout=3)
