用python做测试--Python实现远程性能监控（1）
在性能测试中，监控被测试服务器的性能指标是个重要的工作，包括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,实现了远程发送命令和获取结果。
#!/usr/bin/env python# -*- coding: UTF-8 -*-import time,sys,logging,traceback,telnetlib,socketclass TelnetAction: def __init__(self,host,account,accountPasswd,RootPasswd=""): self.log=logging.getLogger() self.host=host self.account=account self.accountPasswd=accountPasswd self.RootPasswd=RootPasswd self.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.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.prompt) self.child.write('bash'+'\n') self.expand_expect(self.prompt) self.log.info("login host "+self.host+" successfully") return True except: 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.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): return self.child.read_eager() 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 logout(self): self.child.close()后继内容待续。。。。。。