def check_result(result, success_message):
'''
Function to check result of a pexpect operation.
Accepts a success message.
'''
if result == 0:
logger.debug('Test passed: %s' % success_message)
return True
elif result == 1:
logger.warning('EOF - Test failed')
return False
elif result == 2:
logger.warning(' Timed out - Test failed')
return False
else:
logger.warning(' Generic - Test failed')
return False
python类EOF的实例源码
def readline(self, size=-1):
'''This reads and returns one entire line. The newline at the end of
line is returned as part of the string, unless the file ends without a
newline. An empty string is returned if EOF is encountered immediately.
This looks for a newline as a CR/LF pair (\\r\\n) even on UNIX because
this is what the pseudotty device returns. So contrary to what you may
expect you will receive newlines as \\r\\n.
If the size argument is 0 then an empty string is returned. In all
other cases the size argument is ignored, which is not standard
behavior for a file-like object. '''
if size == 0:
return self.string_type()
# delimiter default is EOF
index = self.expect([self.crlf, self.delimiter])
if index == 0:
return self.before + self.crlf
else:
return self.before
def __init__(self, strings):
'''This creates an instance of searcher_string. This argument 'strings'
may be a list; a sequence of strings; or the EOF or TIMEOUT types. '''
self.eof_index = -1
self.timeout_index = -1
self._strings = []
for n, s in enumerate(strings):
if s is EOF:
self.eof_index = n
continue
if s is TIMEOUT:
self.timeout_index = n
continue
self._strings.append((n, s))
def __init__(self, patterns):
'''This creates an instance that searches for 'patterns' Where
'patterns' may be a list or other sequence of compiled regular
expressions, or the EOF or TIMEOUT types.'''
self.eof_index = -1
self.timeout_index = -1
self._searches = []
for n, s in zip(list(range(len(patterns))), patterns):
if s is EOF:
self.eof_index = n
continue
if s is TIMEOUT:
self.timeout_index = n
continue
self._searches.append((n, s))
def __str__(self):
'''This returns a human-readable string that represents the state of
the object.'''
#ss = [(n, ' %d: re.compile("%s")' %
# (n, repr(s.pattern))) for n, s in self._searches]
ss = list()
for n, s in self._searches:
try:
ss.append((n, ' %d: re.compile("%s")' % (n, s.pattern)))
except UnicodeEncodeError:
# for test cases that display __str__ of searches, dont throw
# another exception just because stdout is ascii-only, using
# repr()
ss.append((n, ' %d: re.compile(%r)' % (n, s.pattern)))
ss.append((-1, 'searcher_re:'))
if self.eof_index >= 0:
ss.append((self.eof_index, ' %d: EOF' % self.eof_index))
if self.timeout_index >= 0:
ss.append((self.timeout_index, ' %d: TIMEOUT' %
self.timeout_index))
ss.sort()
ss = list(zip(*ss))[1]
return '\n'.join(ss)
def goToLine(self, fh, linenumber):
"""
Go to 'linenumber' of a huge text file in an (memory-)efficient way.
"""
if linenumber < 1:
raise IOError(
"Specified linenumber '%d' is smaller than 1." % linenumber
)
fh.seek(0, os.SEEK_SET)
# Skip lines until desired line is reached.
for _ in range(0, linenumber - 1):
read = fh.readline()
if read == "":
# Empty string represents EOF.
raise OutOfScopeException(msg="goToLine error: ",
line=linenumber)
def poll(self, timeout=None):
if self.pending_data:
return True
if self.done:
raise EOFError('Child process is done')
if timeout is None:
timeout = 0
while True:
try:
self.child.expect(r'.', timeout=timeout)
except pexpect.TIMEOUT:
return True if self.pending_data else False
except pexpect.EOF:
self.pending_data += self.child.before
self.done = True
if self.pending_data:
return True
raise EOFError('Child process is done')
else:
self.pending_data += self.child.before + self.child.after
def _execute(self, cmd, **kwargs):
"""
Execute a command on a remote host.
Parameters
----------
cmd : string
Command to be executed on remote host.
kwargs : keywords
Options to pass to subprocess.Popen.
Returns
-------
proc : Popen subprocess
Subprocess used to run remote command.
"""
template = 'ssh {login} -T -o ControlPath={socket} << EOF\n{cmd}\nEOF'
config = dict(self._subprocess_config)
config.update(kwargs)
return run_in_subprocess(template.format(login=self._login_info,
socket=self._socket_path,
cmd=cmd),
check_output=True,
**config)
def test_deploy():
directory = mktree(FORGE_YAML + APP, MANGLE=MANGLE)
os.environ["FORGE_PROFILE"] = "dev"
forge = launch(directory, "forge deploy")
forge.expect('built')
forge.expect('forgetest/Dockerfile')
forge.expect('pushed')
forge.expect('forgetest-[0-9-]+:')
forge.expect('rendered')
forge.expect('service/forgetest-[0-9-]+')
forge.expect('deployment/forgetest-[0-9-]+')
forge.expect('deployed')
forge.expect('forgetest-[0-9-]+')
forge.expect(pexpect.EOF)
assert forge.wait() == 0
for sub in ("forgetest", "forgetest/subdir"):
forge = launch(os.path.join(directory, "forgetest/subdir"), "forge deploy")
forge.expect('rendered')
forge.expect('service/forgetest-[0-9-]+')
forge.expect('deployment/forgetest-[0-9-]+')
forge.expect('deployed')
forge.expect('forgetest-[0-9-]+')
forge.expect(pexpect.EOF)
assert forge.wait() == 0
def do_test_rebuilder(tree, path):
directory = mktree(FORGE_YAML + tree, MANGLE=MANGLE)
forge = launch(directory, "forge build containers")
forge.expect(pexpect.EOF)
assert forge.wait() == 0
assert run_image(directory).strip() == "hello"
with open(os.path.join(directory, path), "write") as f:
f.write('print("goodbye")\n')
forge = launch(directory, "forge build containers")
forge.expect(pexpect.EOF)
assert forge.wait() == 0
assert run_image(directory).strip() == "goodbye"
forge = launch(directory, "forge clean")
forge.expect("docker kill ")
forge.expect(pexpect.EOF)
assert forge.wait() == 0
def halt(self):
self.login()
self.child.send("halt\n")
try:
# Wait for text confirming the halt, or EOF
self.child.expect("(The operating system has halted)|(entering state S5)", timeout = 60)
except pexpect.EOF:
# Didn't see the text but got an EOF; that's OK.
print "EOF"
except pexpect.TIMEOUT, e:
# This is unexpected but mostly harmless
print "timeout waiting for halt confirmation:", e
# Calling this directly is deprecated, use Anita.login()
def testCommands(self):
p = pexpect.spawn(self.ravelCmd)
p.expect("ravel>")
p.sendline("exit")
p.expect(pexpect.EOF)
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 exit(self):
"""Send Ctrl + D to exit.
"""
self.cli.sendcontrol('d')
self.cli.expect(pexpect.EOF)
def run(self, cmd):
ssh_newkey = 'Are you sure you want to continue connecting'
child = pexpect.spawn('ssh -l %s -p %s %s %s'%(self._user, self._port, self._host, cmd))
i = child.expect([pexpect.TIMEOUT, ssh_newkey, 'password: '])
if i == 0: #timeout
raise Exception('ssh timeout')
if i == 1: #SSH does not have the public key. Just accept it.
child.sendline('yes')
child.expect('password: ')
i = child.expect([pexpect.TIMEOUT, 'password: '])
if i == 0: #timeout
raise Exception('ssh timeout')
child.sendline(self._pwd)
child.expect(pexpect.EOF)
return child.before
def logout (self):
'''Sends exit to the remote shell.
If there are stopped jobs then this automatically sends exit twice.
'''
self.sendline("exit")
index = self.expect([EOF, "(?i)there are stopped jobs"])
if index==1:
self.sendline("exit")
self.expect(EOF)
self.close()
