剖析 Pexpect（轉）

概述

Pexpect 是 Don Libes 的 Expect 語言的一個 Python 實現，是一個用來啓動子程序，並使用正則表達式對程序輸出作出特定響應，以此實現與其自動交互的 Python 模塊。 Pexpect 的使用範圍很廣，能夠用來實現與 ssh、ftp 、telnet 等程序的自動交互；能夠用來自動複製軟件安裝包並在不一樣機器自動安裝；還能夠用來實現軟件測試中與命令行交互的自動化。

下載

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)

 

依賴

• Python 版本 2.4 或者 2.5
• pty module ，pty 是任何 Posix 系統標準庫的一部分

因爲其依賴 pty module ，因此 Pexpect 還不能在 Windows 的標準 python 環境中執行，若是想在 Windows 平臺使用，可使用在 Windows 中運行 Cygwin 作爲替代方案。

遵循 MIT 許可證

根據 Wiki 對 MIT License 的介紹「該模塊被受權人有權利使用、複製、修改、合併、出版發行、散佈、再受權及販售軟件及軟件的副本。被受權人可根據程序的須要修改受權條款爲適當的內容。 在軟件和軟件的全部副本中都必須包含版權聲明和許可聲明。」

 

回頁首

Pexpect 提供的 run() 函數：

清單 1. run() 的定義

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

 

函數 run 能夠用來運行命令，其做用與 Python os 模塊中 system() 函數類似。run() 是經過 Pexpect 類實現的。

若是命令的路徑沒有徹底給出，則 run 會使用 which 命令嘗試搜索命令的路徑 。

清單 2. 使用 run()執行 svn 命令

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

 

與 os.system() 不一樣的是，使用 run() 能夠方便地同時得到命令的輸出結果與命令的退出狀態 。

清單 3. run() 的返回值

from pexpect import * (command_output, exitstatus) = run ('ls -l /bin', withexitstatus=1)

 

command_out 中保存的就是 /bin 目錄下的內容。

 

回頁首

Pexpect 提供的 spawn() 類：

使用 Pexpect 啓動子程序

清單 4. spawn 的構造函數

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

 

spawn是Pexpect模塊主要的類，用以實現啓動子程序，它有豐富的方法與子程序交互從而實現用戶對子程序的控制。它主要使用 pty.fork() 生成子進程，並調用 exec() 系列函數執行 command 參數的內容。

能夠這樣使用：

清單 5. spawn() 使用示例

child = pexpect.spawn ('/usr/bin/ftp') #執行ftp客戶端命令 child = pexpect.spawn ('/usr/bin/ssh user@example.com') #使用ssh登陸目標機器 child = pexpect.spawn ('ls -latr /tmp') #顯示 /tmp目錄內容

 

當子程序須要參數時，還可使用一個參數的列表：

清單 6. 參數列表示例

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

 

在構造函數中，maxread 屬性指定了 Pexpect 對象試圖從 tty 一次讀取的最大字節數，它的默認值是2000字節 。

因爲須要實現不斷匹配子程序輸出， searchwindowsize 指定了從輸入緩衝區中進行模式匹配的位置，默認從開始匹配。

logfile 參數指定了 Pexpect 產生的日誌的記錄位置。

例如：

清單 7. 記錄日誌

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

 

還能夠將日誌指向標準輸出：

清單 8. 將日誌指向標準輸出

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

 

若是不須要記錄向子程序輸入的日誌，只記錄子程序的輸出，可使用：

清單 9. 記錄輸出日誌

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

 

使用 Pexpect 控制子程序

爲了控制子程序，等待子程序產生特定輸出，作出特定的響應，可使用 expect 方法。

清單 10. expect() 定義

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

 

在參數中： pattern 能夠是正則表達式， pexpect.EOF ， pexpect.TIMEOUT ，或者由這些元素組成的列表。須要注意的是，當 pattern 的類型是一個列表時，且子程序輸出結果中不止一個被匹配成功，則匹配返回的結果是緩衝區中最早出現的那個元素，或者是列表中最左邊的元素。使用 timeout 能夠指定等待結果的超時時間 ，該時間以秒爲單位。當超過預訂時間時， expect 匹配到pexpect.TIMEOUT。

若是難以估算程序運行的時間，可使用循環使其屢次等待直至等待運行結束：

清單 11. 使用循環

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

 

expect() 在執行中可能會拋出兩種類型的異常分別是 EOF and TIMEOUF，其中 EOF 一般表明子程序的退出， TIMEOUT 表明在等待目標正則表達式中出現了超時。

清單 12. 使用並捕獲異常

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()

 

此時能夠將這兩種異常放入expect等待的目標列表中：

清單 13. 避免異常

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()

 

expect 不斷從讀入緩衝區中匹配目標正則表達式，當匹配結束時 pexpect 的 before 成員中保存了緩衝區中匹配成功處以前的內容， pexpect 的 after 成員保存的是緩衝區中與目標正則表達式相匹配的內容。

清單 14. 打印 before 成員的內容

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

 

此時 child.before 保存的就是在根目錄下執行 ls 命令的結果。

清單 15. send 系列函數

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

 

這些方法用來向子程序發送命令，模擬輸入命令的行爲。 與 send() 不一樣的是 sendline() 會額外輸入一個回車符 ，更加適合用來模擬對子程序進行輸入命令的操做。 當須要模擬發送 「Ctrl+c」 的行爲時，還可使用 sendcontrol() 發送控制字符。

清單 16. 發送 ctrl+c

child.sendcontrol('c')

 

因爲 send() 系列函數向子程序發送的命令會在終端顯示，因此也會在子程序的輸入緩衝區中出現，所以不建議使用 expect 匹配最近一次 sendline() 中包含的字符。不然可能會在形成不但願的匹配結果。

清單 17. interact() 定義

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

 

Pexpect還能夠調用 interact() 讓出控制權，用戶能夠繼續當前的會話控制子程序。用戶能夠敲入特定的退出字符跳出，其默認值爲「^]」 。

下面展現一個使用Pexpect和ftp交互的實例。

清單 18. ftp 交互的實例：

# 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')

 

該程序與 ftp 作交互，登陸到 ftp.openbsd.org ，當提述輸入登陸名稱和密碼時輸入默認用戶名和密碼，當出現 「ftp>」 這一提示符時切換到 pub/OpenBSD 目錄並下載 README 這一文件。

如下實例是上述方法的綜合應用，用來創建一個到遠程服務器的 telnet 鏈接，並返回保存該鏈接的 pexpect 對象。

清單 19. 登陸函數：

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

 

 

回頁首

Pxssh 類的使用：

Pxssh 作爲 pexpect 的派生類能夠用來創建一個 ssh 鏈接，它相比其基類增長了以下方法：

login() 創建到目標機器的ssh鏈接 ；

losuckgout() 釋放該鏈接 ；

prompt() 等待提示符，一般用於等待命令執行結束。

下面的示例鏈接到一個遠程服務器，執行命令並打印命令執行結果。

該程序首先接受用戶輸入用戶名和密碼，login 函數返回一個 pxssh 對象的連接，而後調用 sendline() 分別輸入 「uptime」、「ls」 等命令並打印命令輸出結果。

清單 20. pxssh 示例

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)

 

 

回頁首

Pexpect 使用中須要注意的問題：

spawn() 參數的限制

在使用spawn執行命令時應該注意，Pexpect 並不與 shell 的元字符例如重定向符號 > 、>> 、管道 | ，還有通配符 * 等作交互，因此當想運行一個帶有管道的命令時必須另外啓動一個 shell ，爲了使代碼清晰，如下示例使用了參數列表例如：

清單 21. 啓動新的 shell 執行命令

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

 

與線程共同工做

Perl 也有 expect 的模塊 Expect-1.21，可是 perl 的該模塊在某些操做系統例如 fedora 9 或者 AIX 5 中不支持在線程中啓動程序執行如下實例試圖利用多線同時程登陸到兩臺機器進行操做，不使用線程直接調用時 sub1() 函數能夠正常工做，可是使用線程時在 fedora9 和 AIX 5 中都不能正常運行。

清單 22. perl 使用 expect 因爲線程和 expect 共同使用致使不能正常工做的程序

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();

 

Pexpect 則沒有這樣的問題，可使用多線程並在線程中啓動程序運行。可是在某些操做系統如 fedora9 中不能夠在線程之間傳遞 Pexpect 對象。

清單 使用 Pexpect 在線程中啓動控制子程序

請參見實例

對正則表達式的支持

在使用 expect() 時，因爲 Pexpect 是不斷從緩衝區中匹配，若是想匹配行尾不能使用 「$」 ，只能使用 「\r\n」表明一行的結束。 另外其只能獲得最小匹配的結果，而不是進行貪婪匹配，例如 child.expect ('.+') 只能匹配到一個字符。

 

回頁首

應用實例：

在實際系統管理員的任務中，有時須要同時管理多臺機器，這個示例程序被用來自動編譯並安裝新的內核版本，並重啓。它使用多線程，每一個線程都創建一個到遠程機器的 telnet 鏈接並執行相關命令。 該示例會使用上文中的登陸函數。

清單 23. 管理多臺機器示例

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[i] = RefreshKernelThreadClass(linux_server_list[i], same_user,same_passwd) thread_list[i].start() # wait the threads finish for i in range (len(linux_server_list)): thread_list[i].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")