用python做自动化测试--Python实现远程性能监控
2016-07-25 23:11
656 查看
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()
telnetoperate.py代码说明:
1. __init__(self,host,prompt,account,accountPasswd,RootPasswd="")
这里用到了多个登陆账号(account,root),原因是我们的机器开始不能直接root登陆,需要先用普通用户登陆,才能切换到root账号,所以这里出现了account, rootPasswd这2个参数,如果你的机器可以直接root账号登陆,或者你不需要切换到root账号,可以就用account, accountPasswd就可以了。
prompt是命令行提示符,机器配置不一样,可能是$或者#,用来判断一个命令执行是否完成。
2. send_command(self,command,sleeptime=0.5)
这里sleeptime=0.5是为了避免很多机器性能不好,命令执行比较慢,命令还没返回,会导致获取命令后的结果失败。如果你嫌这样太慢了,可以调用的时候send_command(command,0)
process_is_running(
process_name
监控远程机器:
#simplemonitor.py
#!/usr/bin/env python
#coding=utf-8
import time
import telnetoperate
remote_server=telnetoperate.TelnetAction("192.168.23.235","#","user","passwd123")
#get cpu information
cpu=remote_server.get_output("sar 1 1 |tail -1")
memory=remote_server.get_output("top | head -5 |grep -i memory")
io=remote_server.get_output("iostat -x 1 2|grep -v '^$' |grep -vi 'dev'")
这样在任何一台机器上就可以实现监控远程多个机器了,信息集中化管理,方便进一步分析。如果你想cpu, memory, io独立的监控,可以多线程或者起多个监控进程,在多线程中需要注意的时候,必须对每个监控实例建立一个telnet连接,get_output是从telnet 建立的socket里面去获取数据,如果多个监控实例用同一个socket会导致数据混乱。
在性能测试中,监控被测试服务器的性能指标是个重要的工作,包括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()
telnetoperate.py代码说明:
1. __init__(self,host,prompt,account,accountPasswd,RootPasswd="")
这里用到了多个登陆账号(account,root),原因是我们的机器开始不能直接root登陆,需要先用普通用户登陆,才能切换到root账号,所以这里出现了account, rootPasswd这2个参数,如果你的机器可以直接root账号登陆,或者你不需要切换到root账号,可以就用account, accountPasswd就可以了。
prompt是命令行提示符,机器配置不一样,可能是$或者#,用来判断一个命令执行是否完成。
2. send_command(self,command,sleeptime=0.5)
这里sleeptime=0.5是为了避免很多机器性能不好,命令执行比较慢,命令还没返回,会导致获取命令后的结果失败。如果你嫌这样太慢了,可以调用的时候send_command(command,0)
process_is_running(
process_name
监控远程机器:
#simplemonitor.py
#!/usr/bin/env python
#coding=utf-8
import time
import telnetoperate
remote_server=telnetoperate.TelnetAction("192.168.23.235","#","user","passwd123")
#get cpu information
cpu=remote_server.get_output("sar 1 1 |tail -1")
memory=remote_server.get_output("top | head -5 |grep -i memory")
io=remote_server.get_output("iostat -x 1 2|grep -v '^$' |grep -vi 'dev'")
这样在任何一台机器上就可以实现监控远程多个机器了,信息集中化管理,方便进一步分析。如果你想cpu, memory, io独立的监控,可以多线程或者起多个监控进程,在多线程中需要注意的时候,必须对每个监控实例建立一个telnet连接,get_output是从telnet 建立的socket里面去获取数据,如果多个监控实例用同一个socket会导致数据混乱。
相关文章推荐
- Python json模块介绍
- Python之路:爬虫之urllib库的基本使用和高级使用
- Python爬虫学习系列教程-----------爬虫系列 你值的收藏
- python脚本爬取今日百度热点新闻
- python中静态方法和类方法的区别
- BeautifulSoup4的初步使用
- 解决python读取中文文件乱码问题
- python
- python 学习 (七) 循环嵌套
- python(二) N的阶乘对M取余
- Python中的分片
- python笔记---需求文件requirements.txt的创建及使用
- python基础 - 文件读写
- windows64 配置numpy,scipy,theano
- python函数式编程
- Python 添加自定义模块
- Python 添加自定义模块
- python——基本数据类型+常用操作符+循环
- Python语言_思维导图
- 【python】choice函数