Python Pexpect 的使用

lixiaolong

下载地址：https://pypi.python.org/pypi/pexpect/
官方文档：http://pexpect.readthedocs.org/en/latest/

概述
[size=1.166em]Pexpect 是 Don Libes 的 Expect 语言的一个 Python 实现，是一个用来启动子程序，并使用正则表达式对程序输出做出特定响应，以此实现与其自动交互的 Python 模块。 Pexpect 的使用范围很广，可以用来实现与 ssh、ftp 、telnet 等程序的自动交互；可以用来自动复制软件安装包并在不同机器自动安装；还可以用来实现软件测试中与命令行交互的自动化。
下载
[size=1.166em]Pexpect 可以从 SourceForge 网站下载。 本文介绍的示例使用的是 2.3 版本，如不说明测试环境，默认运行操作系统为 fedora 9 并使用 Python 2.5 。
安装


download pexpect-2.3.tar.gz
tar zxvf pexpect-2.3.tar.g
cd pexpect-2.3
python setup.py install  (do this as root)




run(command,timeout=-1,withexitstatus=False,events=None,\
extra_args=None,logfile=None, cwd=None, env=None)


from pexpect import *
run ("svn ci -m 'automatic commit' my_file.py")


from pexpect import *
(command_output, exitstatus) = run ('ls -l /bin', withexitstatus=1)
[size=1.166em]command_out 中保存的就是 /bin 目录下的内容。

Pexpect 提供的 spawn() 类：

class spawn:
def __init__(self,command,args=[],timeout=30,maxread=2000,\
searchwindowsize=None, logfile=None, cwd=None, env=None)


child = pexpect.spawn ('/usr/bin/ftp') #执行ftp客户端命令
child = pexpect.spawn ('/usr/bin/ssh user@example.com') #使用ssh登录目标机器
child = pexpect.spawn ('ls -latr /tmp') #显示 /tmp目录内容


child = pexpect.spawn ('/usr/bin/ftp', [])
child = pexpect.spawn ('/usr/bin/ssh', ['user@example.com'])
child = pexpect.spawn ('ls', ['-latr', '/tmp'])


child = pexpect.spawn('some_command')
fout = file('mylog.txt','w')
child.logfile = fout


child = pexpect.spawn('some_command')
child.logfile = sys.stdout


child = pexpect.spawn('some_command')
child.logfile_send = sys.stdout

expect(self, pattern, timeout=-1, searchwindowsize=None)


while True:
index = child.expect(["suc","fail",pexpect.TIMEOUT])
if index == 0:
break
elif index == 1:
return False
elif index == 2:
pass        #continue to wait


try:
index = pexpect (['good', 'bad'])
if index == 0:
do_something()
elif index == 1:
do_something_else()
except EOF:
do_some_other_thing()
except TIMEOUT:
do_something_completely_different()


index = p.expect (['good', 'bad', pexpect.EOF, pexpect.TIMEOUT])
if index == 0:
do_something()
elif index == 1:
do_something_else()
elif index == 2:
do_some_other_thing()
elif index == 3:
do_something_completely_different()


child = pexpect.spawn('/bin/ls /')
child.expect (pexpect.EOF)
print child.before


send(self, s)
sendline(self, s='')
sendcontrol(self, char)


child.sendcontrol('c')


interact(self, escape_character = chr(29), input_filter = None, output_filter = None)


# This connects to the openbsd ftp site and
# downloads the README file.
import pexpect
child = pexpect.spawn ('ftp ftp.openbsd.org')
child.expect ('Name .*: ')
child.sendline ('anonymous')
child.expect ('Password:')
child.sendline ('noah@example.com')
child.expect ('ftp> ')
child.sendline ('cd pub/OpenBSD')
child.expect('ftp> ')
child.sendline ('get README')
child.expect('ftp> ')
child.sendline ('bye')


import re,sys,os
from pexpect import *

def telnet_login(server,user, passwd,shell_prompt= “#|->”):  
"""  
@summary: This logs the user into the given server.
It uses the 'shell_prompt' to try to find the prompt right after login.
When it finds the prompt it immediately tries to reset the prompt to '#UNIQUEPROMPT#'
more easily matched.  
@return: If Login successfully ,It will return a pexpect object   
@raise exception: RuntimeError will be raised when the cmd telnet failed or the user
and passwd do not match  
@attention:1. shell_prompt should not include '$',on some server, after sendline
(passwd)   the pexpect object will read a '$'.  
2.sometimes the server's output before its shell prompt will contain '#' or
'->'  So the caller should kindly assign the shell prompt  
"""  
if not server or not user \  
or not passwd or not shell_prompt:  
raise RuntimeError, "You entered empty parameter for telnet_login "  

child = pexpect.spawn('telnet %s' % server)  
child.logfile_read = sys.stdout  
index = child.expect (['(?i)login:', '(?i)username', '(?i)Unknown host'])  
if index == 2:  
raise RuntimeError, 'unknown machine_name' + server  
child.sendline (user)  
child.expect ('(?i)password:')  
child.logfile_read = None  # To turn off log
child.sendline (passwd)  

while True:  
index = child.expect([pexpect.TIMEOUT,shell_prompt])  
child.logfile_read = sys.stdout  
if index == 0:  
if re.search('an invalid login', child.before):  
raise RuntimeError, 'You entered an invalid login name or password.'
elif index == 1:  
break      
child.logfile_read = sys.stdout # To tun on log again
child.sendline(“PS1=#UNIQUEPROMPT#”)  
#This is very crucial to wait for PS1 has been modified successfully  
#child.expect(“#UNIQUEPROMPT#”)  
child.expect("%s.+%s" % (“#UNIQUEPROMPT#”, “#UNIQUEPROMPT#”))  
return child



import pxssh  
import getpass  
try:
s = pxssh.pxssh()
hostname = raw_input('hostname: ')  
username = raw_input('username: ')  
password = getpass.getpass('password: ')  
s.login (hostname, username, password)  
s.sendline ('uptime')  # run a command  
s.prompt()             # match the prompt  
print s.before         # print everything before the propt.  
s.sendline ('ls -l')  
s.prompt()
print s.before  
s.sendline ('df')  
s.prompt()
print s.before  
s.logout()
except pxssh.ExceptionPxssh, e:  
print "pxssh failed on login."
print str(e)



shell_cmd = 'ls -l | grep LOG > log_list.txt'
child = pexpect.spawn('/bin/bash', ['-c', shell_cmd])
child.expect(pexpect.EOF)

use threads;
use Expect;  
$timeout = 5;  
my $thr = threads->create(\&sub1(first_server));  
my $thr2 = threads->create(\&sub1(second_server));  
sub sub1  
{  
my $exp = new Expect;  
$exp -> raw_pty(1);  
$exp -> spawn ("telnet",$_[0]) or die "cannot access telnet";
$exp -> expect ( $timeout, -re=>'[Ll]ogin:' );  
$exp -> send ( "user\n");  
$exp -> expect ( $timeout, -re=>'[Pp]assword:' );  
$exp -> send ( "password\n" );  
$exp -> expect ( $timeout, -re=>" #" );  
$exp -> send ( "date\n" );
$exp -> expect ( $timeout, -re=>'\w\w\w \w\w\w \d{1,2}
\d\d:\d\d:\d\d \w\w\w \d\d\d\d');  
$localtime=$exp->match();
print "\tThe first server’s time is : $localtime\n";  
$exp -> soft_close ();
}  
print "This is the main thread!";  
$thr->join();
$thr2->join();
在使用 expect() 时，由于 Pexpect 是不断从缓冲区中匹配，如果想匹配行尾不能使用 “$” ，只能使用 “\r\n”代表一行的结束。 另外其只能得到最小匹配的结果，而不是进行贪婪匹配，例如 child.expect ('.+') 只能匹配到一个字符。



import sys,os  
from Login import *
PROMPT = “#UNIQUEPROMPT#”
class RefreshKernelThreadClass(threading.Thread):  
""" The thread to downLoad the kernel and install it on a new server  """
def __init__(self,server_name,user,passwd):  
threading.Thread.__init__(self)  
self.server_name_ = server_name  
self.user_ = user  
self.passwd_ = passwd  
self.result_ = [] # the result information of the thread

def run(self):  
self.setName(self.server_name_) # set the name of thread

try:  
#call the telnet_login to access the server through telnet
child = telnet_login(self.server_name_,self.user_, self.passwd_)  
except RuntimeError,ex:  
info = "telnet to machine %s failed with reason %s" % (self.server_name_, ex)
self.result_.=(False, self.server_name_+info)
return self.result_   

child.sendline(' cd ~/Download/dw_test && \
wget http://www.kernel.org/pub/linux/kernel/v2.6/linux-2.6.28.tar.gz && \
tar zxvf linux-2.6.28.tar.gz && \
cd linux-2.6.28 \
&& make mrproper && make allyesconfig and make -j 4 && make modules && \
make modules install && make install')  

# wail these commands finish
while True:
index = child.expect([PROMPT,pexpect.TIMEOUT,pexpect.EOF])  
if index == 0:
break
elif index == 1:
pass
elif index ==2 :
self.result_=(False,'Sub process exit abnormally ')
return False

# reboot the server
child.sendline('shutdown -Fr')  
child.expect('\r\n')  
retry_times = 10
while retry_times > 0:   
index_shutdown = child.expect(["Unmounting the file systems",
pexpect.EOF,  
pexpect.TIMEOUT])  
if index_shutdown == 0 or index_shutdown == 1 :  
break  
elif index_shutdown == 2:
retry_times = retry_times-1
if retry_times == 0:  
self.result_=(False,'Cannot shutdown ')  
return self.result_


def refresh_kernel(linux_server_list,same_user,same_passwd):  
"""  
@summary: The function is used to work on different linux servers to download
the same version linux kernel, conpile them and reboot all these servers  
To keep it simple we use the same user id and password on these servers   
"""  
if not type(linux_server_list) == list:
return (False,"Param %s Error!"%linux_server_list)  

if same_user is None or same_passwd is None or not  
type(same_user)== str or not type(same_passwd) == str:  
return (False,"Param Error!")

thread_list = []
# start threads to execute command on the remote servers
for i in range (len(linux_server_list)):  
thread_list = RefreshKernelThreadClass(linux_server_list,
same_user,same_passwd)
thread_list.start()

# wait the threads finish
for i in range (len(linux_server_list)):  
thread_list.join()
# validate the result
for i in range (len(linux_server_list)):  
if thread_list[0].result_[0] == False:  
return False
else:
return True

if __name__ == "__main__":
refresh_kernel(server_list,"test_user","test_passwd")


转载自：http://www.ibm.com/developerworks/cn/linux/l-cn-pexpect1/