用python做自動化測試--Python實現遠程性能監控

http://blog.csdn.net/powerccna/article/details/8044222

在性能測試中，監控被測試服務器的性能指標是個重要的工作，包括CPU/Memory/IO/Network，但大多數人估計都是直接在被測試服務器的運行監控程序。我們開始也是這樣做的。但這樣做帶來一個問題是，測試人員需要在每台被測試服務器上部署監控程序，增加了部署的工作量，而且經常因為Python版本的問題，有些模塊不兼容，或者第三方模塊需要再次安裝。

       改進性能測試監控工具：

1. 能遠程監控被測試服務器，這樣測試人員就不需要在每個被測試機器安裝監控工具了。

         2. 被測試服務器上不需要安裝agent，監控結果能取到本地。

        3. 本地服務器上的python模塊和兼容性問題，可以通過Python  virtualenv解決，每個測試人員維護自己的一套Python環境。

 

Google了下，找到了pymeter（thttp://pymeter.sourceforge.net/）， 看了下源代碼，很多工作還沒完成，但這個思路和我的是一樣的。而且我在其他項目中已經實現了遠程發送命令的模塊。 所以不如直接在自己的項目上擴展。

 

遠程發送命令的模塊開始是基於Pexpect(http://www.noah.org/wiki/Pexpect)實現的, Pexpect很強大，它是一個用來啟動子程序，並使用正則表達式對程序輸出做出特定響應，以此實現與其自動交互的 Python 模塊。用他來可以很容易實現telnet,ftp,ssh的操作。 但Pexpect無windows下的版本，這是我拋棄它的原因，無法達到測試工具兼容所有系統的要求。 所以就用telent模塊替換了Pexpect,實現了遠程發送命令和獲取結果。

#file name: telnetoperate.py

#!/usr/bin/env python
#coding=utf-8

import time,sys,logging,traceback,telnetlib,socket


class TelnetAction:
    def __init__(self,host,prompt,account,accountPasswd,RootPasswd=""):
        self.log=logging.getLogger()
        self.host=host
        self.account=account
        self.accountPasswd=accountPasswd
        self.RootPasswd=RootPasswd
        self.possible_prompt = ["#","$"]
        self.prompt=prompt
        self.default_time_out=20
        self.child =None
        self.login()

    def expand_expect(self,expect_list):
        try:
            result=self.child.expect(expect_list,self.default_time_out)
        except EOFError:
            self.log.error("No text was read, please check reason")
        if result[0]==-1:
            self.log.error("Expect result"+str(expect_list)+" don't exist")
        else:
            pass
        return result

    def login(self):
        """Connect to a remote host and login.

        """
        try:
            self.child = telnetlib.Telnet(self.host)
            self.expand_expect(['login:'])
            self.child.write(self.account+ '\n')
            self.expand_expect(['assword:'])
            self.child.write(self.accountPasswd + '\n')
            self.expand_expect(self.possible_prompt)
            self.log.debug("swith to root account on host "+self.host)
            if self.RootPasswd!="":
                self.child.write('su -'+'\n')
                self.expand_expect(['assword:'])
                self.child.write(self.RootPasswd+'\n')
                self.expand_expect(self.possible_prompt)
            #self.child.write('bash'+'\n')
            #self.expand_expect(self.possible_prompt)
            self.child.read_until(self.prompt)
            self.log.info("login host "+self.host+" successfully")
            return True
        except:
            print "Login failed,please check ip address and account/passwd"
            self.log.error("log in host "+self.host+" failed, please check reason")
            return False

    def send_command(self,command,sleeptime=0.5):
        """Run a command on the remote host.

        @param command: Unix command
        @return: Command output
        @rtype: String
        """
        self.log.debug("Starting to execute command: "+command)
        try:
            self.child.write(command + '\n')
            if self.expand_expect(self.possible_prompt)[0]==-1:
                self.log.error("Executed command "+command+" is failed, please check it")
                return False
            else:
                time.sleep(sleeptime)
                self.log.debug("Executed command "+command+" is successful")
                return True
        except socket.error:
            self.log.error("when executed command "+command+" the connection maybe break, reconnect")
            traceback.print_exc()
            for i in range(0,3):
                self.log.error("Telnet session is broken from "+self.host+ ", reconnecting....")
                if self.login():
                    break
            return False

    def get_output(self,time_out=2):
        reponse=self.child.read_until(self.prompt,time_out)
        #print "response:",reponse
        self.log.debug("reponse:"+reponse)
        return  self.__strip_output(reponse)

    def send_atomic_command(self, command):
        self.send_command(command)
        command_output = self.get_output()
        self.logout()
        return command_output

    def process_is_running(self,process_name,output_string):
        self.send_command("ps -ef | grep "+process_name+" | grep -v grep")
        output_list=[output_string]
        if self.expand_expect(output_list)[0]==-1:
            return False
        else:
            return True

    def __strip_output(self, response):
        #Strip everything from the response except the actual command output.

        #split the response into a list of the lines

        lines = response.splitlines()
        self.log.debug("lines:"+str(lines))
        if len(lines)>1:
            #if our command was echoed back, remove it from the output
            if self.prompt in lines[0]:
                lines.pop(0)
            #remove the last element, which is the prompt being displayed again
            lines.pop()
            #append a newline to each line of output
            lines = [item + '\n' for item in lines]
            #join the list back into a string and return it
            return ''.join(lines)
        else:
            self.log.info("The response is blank:"+response)
            return "Null response"

    def logout(self):

        self.child.close()