python压测脚本

⑴ 如何使用python编写测试脚本

1)doctest
使用doctest是一种类似于命令行尝试的方式，用法很简单，如下

复制代码代码如下:

def f(n):
"""
>>> f(1)
1
>>> f(2)
2
"""
print(n)

if __name__ == '__main__':
import doctest
doctest.testmod()

应该来说是足够简单了，另外还有一种方式doctest.testfile(filename)，就是把命令行的方式放在文件里进行测试。

2)unittest
unittest历史悠久，最早可以追溯到上世纪七八十年代了，C++，Java里也都有类似的实现，Python里的实现很简单。
unittest在python里主要的实现方式是TestCase,TestSuite。用法还是例子起步。

复制代码代码如下:

from widget import Widget
import unittest
# 执行测试的类
class WidgetTestCase(unittest.TestCase):
def setUp(self):
self.widget = Widget()
def tearDown(self):
self.widget.dispose()
self.widget = None
def testSize(self):
self.assertEqual(self.widget.getSize(), (40, 40))
def testResize(self):
self.widget.resize(100, 100)
self.assertEqual(self.widget.getSize(), (100, 100))
# 测试
if __name__ == "__main__":
# 构造测试集
suite = unittest.TestSuite()
suite.addTest(WidgetTestCase("testSize"))
suite.addTest(WidgetTestCase("testResize"))

# 执行测试
runner = unittest.TextTestRunner()
runner.run(suite)

简单的说，1>构造TestCase(测试用例)，其中的setup和teardown负责预处理和善后工作。2>构造测试集，添加用例3>执行测试需要说明的是测试方法，在Python中有N多测试函数，主要的有：
TestCase.assert_(expr[, msg])
TestCase.failUnless(expr[, msg])
TestCase.assertTrue(expr[, msg])
TestCase.assertEqual(first, second[, msg])
TestCase.failUnlessEqual(first, second[, msg])
TestCase.assertNotEqual(first, second[, msg])
TestCase.failIfEqual(first, second[, msg])
TestCase.assertAlmostEqual(first, second[, places[, msg]])
TestCase.failUnlessAlmostEqual(first, second[, places[, msg]])
TestCase.assertNotAlmostEqual(first, second[, places[, msg]])
TestCase.failIfAlmostEqual(first, second[, places[, msg]])
TestCase.assertRaises(exception, callable, ...)
TestCase.failUnlessRaises(exception, callable, ...)
TestCase.failIf(expr[, msg])
TestCase.assertFalse(expr[, msg])
TestCase.fail([msg])

⑵ python 运维常用脚本

Python 批量遍历目录文件,并修改访问时间

import os

path = "D:/UASM64/include/"
dirs = os.listdir(path)
temp=[];

for file in dirs:
temp.append(os.path.join(path, file))
for x in temp:
os.utime(x, (1577808000, 1577808000))
Python 实现的自动化服务器管理

import sys
import os
import paramiko

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_cmd(user,passwd,port,userfile,cmd):

def ssh_put(user,passwd,source,target):

while True:
try:
shell=str(input("[Shell] # "))
if (shell == ""):
continue
elif (shell == "exit"):
exit()
elif (shell == "put"):
ssh_put("root","123123","./a.py","/root/a.py")
elif (shell =="cron"):
temp=input("输入一个计划任务: ")
temp1="(crontab -l; echo "+ temp + ") |crontab"
ssh_cmd("root","123123","22","./user_ip.conf",temp1)
elif (shell == "uncron"):
temp=input("输入要删除的计划任务: ")
temp1="crontab -l | grep -v " "+ temp + "|crontab"
ssh_cmd("root","123123","22","./user_ip.conf",temp1)
else:
ssh_cmd("lyshark","123123","22","./user_ip.conf",shell)

遍历目录和文件

import os

def list_all_files(rootdir):
import os
_files = []
list = os.listdir(rootdir) #列出文件夹下所有的目录与文件
for i in range(0,len(list)):
path = os.path.join(rootdir,list[i])
if os.path.isdir(path):
_files.extend(list_all_files(path))
if os.path.isfile(path):
_files.append(path)
return _files

a=list_all_files("C:/Users/LyShark/Desktop/a")
print(a)
python检测指定端口状态

import socket

sk = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
sk.settimeout(1)

for ip in range(0,254):
try:
sk.connect(("192.168.1."+str(ip),443))
print("192.168.1.%d server open "%ip)
except Exception:
print("192.168.1.%d server not open"%ip)

sk.close()

python实现批量执行CMD命令

import sys
import os
import paramiko

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

print("------------------------------> ")
print("使用说明,在当前目录创建ip.txt写入ip地址")
print("------------------------------> ")

user=input("输入用户名:")
passwd=input("输入密码:")
port=input("输入端口:")
cmd=input("输入执行的命令:")

file = open("./ip.txt", "r")
line = file.readlines()

for i in range(len(line)):
print("对IP: %s 执行"%line[i].strip(' '))

python3-实现钉钉报警

import requests
import sys
import json

dingding_url = ' https://oapi.dingtalk.com/robot/send?access_token='

data = {"msgtype": "markdown","markdown": {"title": "监控","text": "apche异常"}}

headers = {'Content-Type':'application/json;charset=UTF-8'}

send_data = json.mps(data).encode('utf-8')
requests.post(url=dingding_url,data=send_data,headers=headers)

import psutil
import requests
import time
import os
import json

monitor_name = set(['httpd','cobblerd']) # 用户指定监控的服务进程名称

proc_dict = {}
proc_name = set() # 系统检测的进程名称
monitor_map = {
'httpd': 'systemctl restart httpd',
'cobblerd': 'systemctl restart cobblerd' # 系统在进程down掉后，自动重启
}

dingding_url = ' https://oapi.dingtalk.com/robot/send?access_token='

while True:
for proc in psutil.process_iter(attrs=['pid','name']):
proc_dict[proc.info['pid']] = proc.info['name']
proc_name.add(proc.info['name'])

判断指定端口是否开放

import socket

port_number = [135,443,80]

for index in port_number:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex((飗.0.0.1', index))
if result == 0:
print("Port %d is open" % index)
else:
print("Port %d is not open" % index)
sock.close()

判断指定端口并且实现钉钉轮询报警

import requests
import sys
import json
import socket
import time

def dingding(title,text):
dingding_url = ' https://oapi.dingtalk.com/robot/send?access_token='
data = {"msgtype": "markdown","markdown": {"title": title,"text": text}}
headers = {'Content-Type':'application/json;charset=UTF-8'}
send_data = json.mps(data).encode('utf-8')
requests.post(url=dingding_url,data=send_data,headers=headers)

def net_scan():
port_number = [80,135,443]
for index in port_number:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex((飗.0.0.1', index))
if result == 0:
print("Port %d is open" % index)
else:
return index
sock.close()

while True:
dingding("Warning",net_scan())
time.sleep(60)

python-实现SSH批量CMD执行命令

import sys
import os
import paramiko

ssh = paramiko.SSHClient()
ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())

def ssh_cmd(user,passwd,port,userfile,cmd):
file = open(userfile, "r")
line = file.readlines()
for i in range(len(line)):
print("对IP: %s 执行"%line[i].strip(' '))
ssh.connect(hostname=line[i].strip(' '),port=port,username=user,password=passwd)
cmd=cmd
stdin, stdout, stderr = ssh.exec_command(cmd)
result = stdout.read()

ssh_cmd("lyshark","123","22","./ip.txt","free -h |grep 'Mem:' |awk '{print $3}'")

用python写一个列举当前目录以及所有子目录下的文件，并打印出绝对路径

import sys
import os

for root,dirs,files in os.walk("C://"):
for name in files:
print(os.path.join(root,name))
os.walk()

按照这样的日期格式（xxxx-xx-xx）每日生成一个文件，例如今天生成的文件为2013-09-23.log， 并且把磁盘的使用情况写到到这个文件中。

import os
import sys
import time

new_time = time.strftime("%Y-%m-%d")
disk_status = os.popen("df -h").readlines()

str1 = ''.join(disk_status)
f = open(new_time+'.log','w')
f.write("%s"%str1)

f.flush()
f.close()

统计出每个IP的访问量有多少？（从日志文件中查找）

import sys

list = []

f = open("/var/log/httpd/access_log","r")
str1 = f.readlines()
f.close()

for i in str1:
ip=i.split()[0]
list.append(ip)

list_num=set(list)

for j in list_num:
num=list.count(j)
print("%s -----> %s" %(num,j))

写个程序，接受用户输入数字，并进行校验，非数字给出错误提示，然后重新等待用户输入。

import tab
import sys

while True:
try:
num=int(input("输入数字：").strip())
for x in range(2,num+1):
for y in range(2,x):
if x % y == 0:
break
else:
print(x)
except ValueError:
print("您输入的不是数字")
except KeyboardInterrupt:
sys.exit(" ")

ps 可以查看进程的内存占用大小，写一个脚本计算一下所有进程所占用内存大小的和。

import sys
import os

list=[]
sum=0

str1=os.popen("ps aux","r").readlines()

for i in str1:
str2=i.split()
new_rss=str2[5]
list.append(new_rss)
for i in list[1:-1]:
num=int(i)
sum=sum+num

print("%s ---> %s"%(list[0],sum))

关于Python 命令行参数argv

import sys

if len(sys.argv) < 2:
print ("没有输入任何参数")
sys.exit()

if sys.argv[1].startswith("-"):
option = sys.argv[1][1:]

利用random生成6位数字加字母随机验证码

import sys
import random

rand=[]

for x in range(6):
y=random.randrange(0,5)
if y == 2 or y == 4:
num=random.randrange(0,9)
rand.append(str(num))
else:
temp=random.randrange(65,91)
c=chr(temp)
rand.append(c)
result="".join(rand)
print(result)

自动化-使用pexpect非交互登陆系统

import pexpect
import sys

ssh = pexpect.spawn('ssh [email protected]')
fout = file('sshlog.txt', 'w')
ssh.logfile = fout

ssh.expect("[email protected]'s password:")

ssh.sendline("密码")
ssh.expect('#')

ssh.sendline('ls /home')
ssh.expect('#')

Python-取系统时间

import sys
import time

time_str = time.strftime("日期：%Y-%m-%d",time.localtime())
print(time_str)

time_str= time.strftime("时间：%H:%M",time.localtime())
print(time_str)

psutil-获取内存使用情况

import sys
import os
import psutil

memory_convent = 1024 * 1024
mem =psutil.virtual_memory()

print("内存容量为:"+str(mem.total/(memory_convent))+"MB ")
print("已使用内存:"+str(mem.used/(memory_convent))+"MB ")
print("可用内存:"+str(mem.total/(memory_convent)-mem.used/(1024*1024))+"MB ")
print("buffer容量:"+str(mem.buffers/( memory_convent ))+"MB ")
print("cache容量:"+str(mem.cached/(memory_convent))+"MB ")

Python-通过SNMP协议监控CPU
注意：被监控的机器上需要支持snmp协议 yum install -y net-snmp*

import os

def getAllitems(host, oid):
sn1 = os.popen('snmpwalk -v 2c -c public ' + host + ' ' + oid + '|grep Raw|grep Cpu|grep -v Kernel').read().split(' ')[:-1]
return sn1

def getDate(host):
items = getAllitems(host, '.1.3.6.1.4.1.2021.11')

if name == ' main ':

Python-通过SNMP协议监控系统负载
注意：被监控的机器上需要支持snmp协议 yum install -y net-snmp*

import os
import sys

def getAllitems(host, oid):
sn1 = os.popen('snmpwalk -v 2c -c public ' + host + ' ' + oid).read().split(' ')
return sn1

def getload(host,loid):
load_oids = Ƈ.3.6.1.4.1.2021.10.1.3.' + str(loid)
return getAllitems(host,load_oids)[0].split(':')[3]

if name == ' main ':

Python-通过SNMP协议监控内存
注意：被监控的机器上需要支持snmp协议 yum install -y net-snmp*

import os

def getAllitems(host, oid):

def getSwapTotal(host):

def getSwapUsed(host):

def getMemTotal(host):

def getMemUsed(host):

if name == ' main ':

Python-通过SNMP协议监控磁盘
注意：被监控的机器上需要支持snmp协议 yum install -y net-snmp*

import re
import os

def getAllitems(host,oid):

def getDate(source,newitem):

def getRealDate(item1,item2,listname):

def caculateDiskUsedRate(host):

if name == ' main ':

Python-通过SNMP协议监控网卡流量
注意：被监控的机器上需要支持snmp协议 yum install -y net-snmp*

import re
import os

def getAllitems(host,oid):
sn1 = os.popen('snmpwalk -v 2c -c public ' + host + ' ' + oid).read().split(' ')[:-1]
return sn1

def getDevices(host):
device_mib = getAllitems(host,'RFC1213-MIB::ifDescr')
device_list = []

def getDate(host,oid):
date_mib = getAllitems(host,oid)[1:]
date = []

if name == ' main ':

Python-实现多级菜单

import os
import sys

ps="[None]->"
ip=["192.168.1.1","192.168.1.2","192.168.1.3"]
flage=1

while True:
ps="[None]->"
temp=input(ps)
if (temp=="test"):
print("test page !!!!")
elif(temp=="user"):
while (flage == 1):
ps="[User]->"
temp1=input(ps)
if(temp1 =="exit"):
flage=0
break
elif(temp1=="show"):
for i in range(len(ip)):
print(i)

Python实现一个没用的东西

import sys

ps="[root@localhost]# "
ip=["192.168.1.1","192.168.1.2","192.168.1.3"]

while True:
temp=input(ps)
temp1=temp.split()

检查各个进程读写的磁盘IO

import sys
import os
import time
import signal
import re

class DiskIO:
def init (self, pname=None, pid=None, reads=0, writes=0):
self.pname = pname
self.pid = pid
self.reads = 0
self.writes = 0

def main():
argc = len(sys.argv)
if argc != 1:
print ("usage: please run this script like [./lyshark.py]")
sys.exit(0)
if os.getuid() != 0:
print ("Error: This script must be run as root")
sys.exit(0)
signal.signal(signal.SIGINT, signal_handler)
os.system('echo 1 > /proc/sys/vm/block_mp')
print ("TASK PID READ WRITE")
while True:
os.system('dmesg -c > /tmp/diskio.log')
l = []
f = open('/tmp/diskio.log', 'r')
line = f.readline()
while line:
m = re.match(
'^(S+)(d+)(d+): (READ|WRITE) block (d+) on (S+)', line)
if m != None:
if not l:
l.append(DiskIO(m.group(1), m.group(2)))
line = f.readline()
continue
found = False
for item in l:
if item.pid == m.group(2):
found = True
if m.group(3) == "READ":
item.reads = item.reads + 1
elif m.group(3) == "WRITE":
item.writes = item.writes + 1
if not found:
l.append(DiskIO(m.group(1), m.group(2)))
line = f.readline()
time.sleep(1)
for item in l:
print ("%-10s %10s %10d %10d" %
(item.pname, item.pid, item.reads, item.writes))
def signal_handler(signal, frame):
os.system('echo 0 > /proc/sys/vm/block_mp')
sys.exit(0)

if name ==" main ":
main()

利用Pexpect实现自动非交互登陆linux

import pexpect
import sys

ssh = pexpect.spawn('ssh [email protected]')
fout = file('sshlog.log', 'w')
ssh.logfile = fout

ssh.expect("[email protected]'s password:")

ssh.sendline("密码")

ssh.expect('#')
ssh.sendline('ls /home')
ssh.expect('#')

利用psutil模块获取系统的各种统计信息

import sys
import psutil
import time
import os

time_str = time.strftime( "%Y-%m-%d", time.localtime( ) )
file_name = "./" + time_str + ".log"

if os.path.exists ( file_name ) == False :
os.mknod( file_name )
handle = open ( file_name , "w" )
else :
handle = open ( file_name , "a" )

if len( sys.argv ) == 1 :
print_type = 1
else :
print_type = 2

def isset ( list_arr , name ) :
if name in list_arr :
return True
else :
return False

print_str = "";

if ( print_type == 1 ) or isset( sys.argv,"mem" ) :
memory_convent = 1024 * 1024
mem = psutil.virtual_memory()
print_str += " 内存状态如下: "
print_str = print_str + " 系统的内存容量为: "+str( mem.total/( memory_convent ) ) + " MB "
print_str = print_str + " 系统的内存以使用容量为: "+str( mem.used/( memory_convent ) ) + " MB "
print_str = print_str + " 系统可用的内存容量为: "+str( mem.total/( memory_convent ) - mem.used/( 1024*1024 )) + "MB "
print_str = print_str + " 内存的buffer容量为: "+str( mem.buffers/( memory_convent ) ) + " MB "
print_str = print_str + " 内存的cache容量为:" +str( mem.cached/( memory_convent ) ) + " MB "

if ( print_type == 1 ) or isset( sys.argv,"cpu" ) :
print_str += " CPU状态如下: "
cpu_status = psutil.cpu_times()
print_str = print_str + " user = " + str( cpu_status.user ) + " "
print_str = print_str + " nice = " + str( cpu_status.nice ) + " "
print_str = print_str + " system = " + str( cpu_status.system ) + " "
print_str = print_str + " idle = " + str ( cpu_status.idle ) + " "
print_str = print_str + " iowait = " + str ( cpu_status.iowait ) + " "
print_str = print_str + " irq = " + str( cpu_status.irq ) + " "
print_str = print_str + " softirq = " + str ( cpu_status.softirq ) + " "
print_str = print_str + " steal = " + str ( cpu_status.steal ) + " "
print_str = print_str + " guest = " + str ( cpu_status.guest ) + " "

if ( print_type == 1 ) or isset ( sys.argv,"disk" ) :
print_str += " 硬盘信息如下: "
disk_status = psutil.disk_partitions()
for item in disk_status :
print_str = print_str + " "+ str( item ) + " "

if ( print_type == 1 ) or isset ( sys.argv,"user" ) :
print_str += " 登录用户信息如下: "
user_status = psutil.users()
for item in user_status :
print_str = print_str + " "+ str( item ) + " "

print_str += "--------------------------------------------------------------- "
print ( print_str )
handle.write( print_str )
handle.close()

import psutil

mem = psutil.virtual_memory()
print mem.total,mem.used,mem
print psutil.swap_memory() # 输出获取SWAP分区信息

cpu = psutil.cpu_stats()
printcpu.interrupts,cpu.ctx_switches

psutil.cpu_times(percpu=True) # 输出每个核心的详细CPU信息
psutil.cpu_times().user # 获取CPU的单项数据 [用户态CPU的数据]
psutil.cpu_count() # 获取CPU逻辑核心数，默认logical=True
psutil.cpu_count(logical=False) # 获取CPU物理核心数

psutil.disk_partitions() # 列出全部的分区信息
psutil.disk_usage('/') # 显示出指定的挂载点情况【字节为单位】
psutil.disk_io_counters() # 磁盘总的IO个数
psutil.disk_io_counters(perdisk=True) # 获取单个分区IO个数

psutil.net_io_counter() 获取网络总的IO，默认参数pernic=False
psutil.net_io_counter(pernic=Ture)获取网络各个网卡的IO

psutil.pids() # 列出所有进程的pid号
p = psutil.Process(2047)
p.name() 列出进程名称
p.exe() 列出进程bin路径
p.cwd() 列出进程工作目录的绝对路径
p.status()进程当前状态[sleep等状态]
p.create_time() 进程创建的时间 [时间戳格式]
p.uids()
p.gids()
p.cputimes() 【进程的CPU时间，包括用户态、内核态】
p.cpu_affinity() # 显示CPU亲缘关系
p.memory_percent() 进程内存利用率
p.meminfo() 进程的RSS、VMS信息
p.io_counters() 进程IO信息，包括读写IO数及字节数
p.connections() 返回打开进程socket的nametples列表
p.num_threads() 进程打开的线程数

import psutil
from subprocess import PIPE
p =psutil.Popen(["/usr/bin/python" ,"-c","print 'helloworld'"],stdout=PIPE)
p.name()
p.username()
p.communicate()
p.cpu_times()

psutil.users() # 显示当前登录的用户，和Linux的who命令差不多

psutil.boot_time() 结果是个UNIX时间戳，下面我们来转换它为标准时间格式，如下：
datetime.datetime.fromtimestamp(psutil.boot_time()) # 得出的结果不是str格式，继续进行转换 datetime.datetime.fromtimestamp(psutil.boot_time()).strftime('%Y-%m-%d%H:%M:%S')

Python生成一个随机密码

import random, string
def GenPassword(length):

if name == ' main ':
print (GenPassword(6))

⑶ Python实现性能自动化测试竟然如此简单

一、思考❓❔

1.什么是性能自动化测试?

2.Python中的性能自动化测试库？

locust库

二、基础操作

1.安装locust

安装成功之后，在cmd控制台将会新增一条命令，可输入如下命令查看：

2.基本用法

三、综合案例演练

1.编写自动化测试脚本

2.使用命令行运行

3.打开web ui界面进行配置

设置并发用户数为10，每5秒创建一个用户

压测过程截图

美轮美奂的压测报告

压测失败详情

下载压测统计数据

下载的压测统计数据csv文件

六、总结

出处：https://www.cnblogs.com/keyou1/

⑷ python开发命令行脚本

工作中会经常需要写一些命令行脚本，如果还是用if,else判断用户输入实在是太丑陋了。这里介绍几个python里的命令行脚本库，可以帮助我们快速开发好用的命令行脚本。

https://docs.python.org/3/library/cmd.html

使用方式是继承Cmd,实现自己的子类。

参数comletekey是自动补全操作，默认值是Tab, 如果不为None 且readline可用的话，命令会自动完成。
这里的readline指的是python实现的 GNU readline 接口（标准python库里没有，Windows系统不支持）。

参数stdin,stdout是输入输出流，默认是sys.stdin,sys.stout。

cmd提供了一个简单的框架，但是功能比较简单，python还有其他的很多第三方库可以用来写命令行程序。

https://www.cnblogs.com/xueweihan/p/12293402.html 这篇文章对比了各个库的功能，贴在这里：

看起来fire是最简单的，来试一下。

fire 则是用一种面向广义对象的方式来玩转命令行，这种对象可以是类、函数、字典、列表等，它更加灵活，也更加简单。你都不需要定义参数类型，fire 会根据输入和参数默认值来自动判断，这无疑进一步简化了实现过程。

以下示例为 fire 实现的 计算器程序：

从上述示例可以看出，fire 提供的方式无疑是最简单、并且最 Pythonic 的了。我们只需关注业务逻辑，而命令行参数的定义则和函数参数的定义融为了一体。

不过，有利自然也有弊，比如 nums 并没有说是什么类型，也就意味着输入字符串'abc'也是合法的，这就意味着一个严格的命令行程序必须在自己的业务逻辑中来对期望的类型进行约束。

⑸ Python - pytest

目录

pytest是Python的单元测试框架，同自带的unittest框架类似，但pytest框架使用起来更简洁，效率更高。

pytest特点

安装

测试

在测试之前要做的准备

我的演示脚本处于这样一个的目录中：

踩坑：你创建的pytest脚本名称中不允许含有 . ，比如 1.简单上手.py ，这样会报错。当然，可以这么写 1-简单上手.py

demo1.py ：

上例中，当我们在执行（就像Python解释器执行普通的Python脚本一样）测试用例的时候， pytest.main(["-s", "demo1.py"]) 中的传参需要是一个元组或者列表（我的pytest是5.2.2版本），之前的版本可能需要这么调用 pytest.main("-s demo1.py") ，传的参数是str的形式，至于你使用哪种，取决于报不报错：

遇到上述报错，就是参数需要一个列表或者元组的形式，而我们使用的是str形式。

上述代码正确的执行结果是这样的：

大致的信息就是告诉我们:

pytest.main(["-s", "demo1.py"])参数说明

除了上述的函数这种写法，也可以有用例类的写法：

用法跟unittest差不多，类名要以 Test 开头，并且其中的用例方法也要以 test 开头，然后执行也一样。

执行结果：

那么，你这个时候可能会问，我记得unittest中有setup和teardown的方法，难道pytest中没有嘛？你怎么提都不提？稳住，答案是有的。

接下来，我们来研究一下pytest中的setup和teardown的用法。

我们知道，在unittest中，setup和teardown可以在每个用例前后执行，也可以在所有的用例集执行前后执行。那么在pytest中，有以下几种情况：

来一一看看各自的用法。

模块级别setup_mole/teardown_mole

执行结果：

类级别的setup_class/teardown_class

执行结果：

类中方法级别的setup_method/teardown_method

执行结果：

函数级别的setup_function/teardown_function

执行结果：

小结

该脚本有多种运行方式，如果处于PyCharm环境，可以使用右键或者点击运行按钮运行，也就是在pytest中的主函数中运行：

也可以在命令行中运行：

这种方式，跟使用Python解释器执行Python脚本没有什么两样。也可以如下面这么执行：

当然，还有一种是使用配置文件运行，来看看怎么用。

在项目的根目录下，我们可以建立一个 pytest.ini 文件，在这个文件中，我们可以实现相关的配置：

那这个配置文件中的各项都是什么意思呢？

首先， pytest.ini 文件必须位于项目的根目录，而且也必须叫做 pytest.ini 。

其他的参数：

OK，来个示例。

首先，（详细目录参考开头的目录结构）在 scripts/test_case_01.py 中：

在 scripts/test_case_dir1/test_case02.py 中：

那么，在不同的目录或者文件中，共有5个用例将被执行，而结果则是两个失败三个成功。来执行验证一下，因为有了配置文件，我们在终端中（前提是在项目的根目录），直接输入 pytest 即可。

由执行结果可以发现， 2 failed, 3 passed ，跟我们的预期一致。

后续执行相关配置都来自配置文件，如果更改，会有相应说明，终端都是直接使用 pytest 执行。

我们知道在unittest中，跳过用例可以用 skip ，那么这同样是适用于pytest。

来看怎么使用：

跳过用例，我们使用 @pytest.mark.skipif(condition, reason) ：

然后将它装饰在需要被跳过用例的的函数上面。

效果如下：

上例执行结果相对详细，因为我们在配置文件中为 addopts 增加了 -v ，之前的示例结果中，没有加！
另外，此时，在输出的控制台中， 还无法打印出 reason 信息，如果需要打印，则可以在配置文件中的 addopts 参数的 -s 变为 -rs ：

如果我们事先知道测试函数会执行失败，但又不想直接跳过，而是希望显示的提示。

Pytest 使用 pytest.mark.xfail 实现预见错误功能：：

需要掌握的必传参数的是：

那么关于预期失败的几种情况需要了解一下：

结果如下：

pytest 使用 x 表示预见的失败（XFAIL）。

如果预见的是失败，但实际运行测试却成功通过，pytest 使用 X 进行标记（XPASS）。

而在预期失败的两种情况中，我们不希望出现预期失败，结果却执行成功了的情况出现，因为跟我们想的不一样嘛，我预期这条用例失败，那这条用例就应该执行失败才对，你虽然执行成功了，但跟我想的不一样，你照样是失败的！

所以，我们需要将预期失败，结果却执行成功了的用例标记为执行失败，可以在 pytest.ini 文件中，加入：

这样就就把上述的情况标记为执行失败了。

pytest身为强大的单元测试框架，那么同样支持DDT数据驱动测试的概念。也就是当对一个测试函数进行测试时，通常会给函数传递多组参数。比如测试账号登陆，我们需要模拟各种千奇百怪的账号密码。

当然，我们可以把这些参数写在测试函数内部进行遍历。不过虽然参数众多，但仍然是一个测试，当某组参数导致断言失败，测试也就终止了。

通过异常捕获，我们可以保证程所有参数完整执行，但要分析测试结果就需要做不少额外的工作。

在 pytest 中，我们有更好的解决方法，就是参数化测试，即每组参数都独立执行一次测试。使用的工具就是 pytest.mark.parametrize(argnames, argvalues) 。

使用就是以装饰器的形式使用。

只有一个参数的测试用例

来看（重要部分）结果：：

可以看到，列表内的每个手机号，都是一条测试用例。

多个参数的测试用例

（重要部分）结果：

可以看到，每一个手机号与每一个验证码都组合一起执行了，这样就执行了4次。那么如果有很多个组合的话，用例数将会更多。我们希望手机号与验证码一一对应组合，也就是只执行两次，怎么搞呢？

在多参数情况下，多个参数名是以 , 分割的字符串。参数值是列表嵌套的形式组成的。

固件（Fixture）是一些函数，pytest 会在执行测试函数之前（或之后）加载运行它们，也称测试夹具。

我们可以利用固件做任何事情，其中最常见的可能就是数据库的初始连接和最后关闭操作。

Pytest 使用 pytest.fixture() 定义固件，下面是最简单的固件，访问主页前必须先登录：

结果：

在之前的示例中，你可能会觉得，这跟之前的setup和teardown的功能也类似呀，但是，fixture相对于setup和teardown来说更灵活。pytest通过 scope 参数来控制固件的使用范围，也就是作用域。

比如之前的login固件，可以指定它的作用域：

很多时候需要在测试前进行预处理（如新建数据库连接），并在测试完成进行清理（关闭数据库连接）。

当有大量重复的这类操作，最佳实践是使用固件来自动化所有预处理和后处理。

Pytest 使用 yield 关键词将固件分为两部分， yield 之前的代码属于预处理，会在测试前执行； yield 之后的代码属于后处理，将在测试完成后执行。

以下测试模拟数据库查询，使用固件来模拟数据库的连接关闭：

结果：

可以看到在两个测试用例执行前后都有预处理和后处理。

pytest中还有非常多的插件供我们使用，我们来介绍几个常用的。

先来看一个重要的，那就是生成测试用例报告。

想要生成测试报告，首先要有下载，才能使用。

下载

如果下载失败，可以使用PyCharm下载，怎么用PyCharm下载这里无需多言了吧。

使用

在配置文件中，添加参数：

效果很不错吧！

没完，看我大招

Allure框架是一个灵活的轻量级多语言测试报告工具，它不仅以web的方式展示了简洁的测试结果，而且允许参与开发过程的每个人从日常执行的测试中最大限度的提取有用信息。
从开发人员（dev，developer）和质量保证人员（QA，Quality Assurance）的角度来看，Allure报告简化了常见缺陷的统计：失败的测试可以分为bug和被中断的测试，还可以配置日志、步骤、fixture、附件、计时、执行 历史 以及与TMS和BUG管理系统集成，所以，通过以上配置，所有负责的开发人员和测试人员可以尽可能的掌握测试信息。
从管理者的角度来看，Allure提供了一个清晰的“大图”，其中包括已覆盖的特性、缺陷聚集的位置、执行时间轴的外观以及许多其他方便的事情。allure的模块化和可扩展性保证了我们总是能够对某些东西进行微调。

少扯点，来看看怎么使用。

Python的pytest中allure下载

但由于这个 allure-pytest 插件生成的测试报告不是 html 类型的，我们还需要使用allure工具再“加工”一下。所以说，我们还需要下载这个allure工具。

allure工具下载

在现在allure工具之前，它依赖Java环境，我们还需要先配置Java环境。

注意，如果你的电脑已经有了Java环境，就无需重新配置了。

配置完了Java环境，我们再来下载allure工具，我这里直接给出了网络云盘链接，你也可以去其他链接中自行下载：

下载并解压好了allure工具包之后，还需要将allure包内的 bin 目录添加到系统的环境变量中。

完事后打开你的终端测试：

返回了版本号说明安装成功。

使用

一般使用allure要经历几个步骤：

来看配置 pytest.ini ：

就是 --alluredir ./report/result 参数。

在终端中输入 pytest 正常执行测试用例即可：

执行完毕后，在项目的根目下，会自动生成一个 report 目录，这个目录下有：

接下来需要使用allure工具来生成HTML报告。

此时我们在终端（如果是windows平台，就是cmd），路径是项目的根目录，执行下面的命令。

PS：我在pycharm中的terminal输入allure提示'allure' 不是内部或外部命令，也不是可运行的程序或批处理文件。但windows的终端没有问题。

命令的意思是，根据 reportresult 目录中的数据（这些数据是运行pytest后产生的）。在 report 目录下新建一个 allure_html 目录，而这个目录内有 index.html 才是最终的allure版本的HTML报告；如果你是重复执行的话，使用 --clean 清除之前的报告。

结果很漂亮：

allure open
默认的，allure报告需要HTTP服务器来打开，一般我们可以通过pycharm来完成，另外一种情况就是通过allure自带的open命令来完成。

allure的其他用法
当然，故事还是没有完！在使用allure生成报告的时候，在编写用例阶段，还可以有一些参数可以使用：

allure.title与allure.description

feature和story

由上图可以看到，不同的用例被分为不同的功能中。

allure.severity

allure.severity 用来标识测试用例或者测试类的级别，分为blocker，critical，normal，minor，trivial5个级别。

severity的默认级别是normal，所以上面的用例5可以不添加装饰器了。

allure.dynamic

在之前，用例的执行顺序是从上到下依次执行：

正如上例的执行顺序是 3 1 2 。

现在，来看看我们如何手动控制多个用例的执行顺序，这里也依赖一个插件。

下载

使用

手动控制用例执行顺序的方法是在给各用例添加一个装饰器：

那么， 现在的执行顺序是 2 1 3 ，按照order指定的排序执行的。

如果有人较劲传个0或者负数啥的，那么它们的排序关系应该是这样的：

失败重试意思是指定某个用例执行失败可以重新运行。

下载

使用

需要在 pytest.ini 文件中， 配置：

给 addopts 字段新增（其他原有保持不变） --reruns=3 字段，这样如果有用例执行失败，则再次执行，尝试3次。

来看示例：

结果：

我们也可以从用例报告中看出重试的结果：

上面演示了用例失败了，然后重新执行多少次都没有成功，这是一种情况。

接下来，来看另一种情况，那就是用例执行失败，重新执行次数内通过了，那么剩余的重新执行的次数将不再执行。

通过 random 模块帮助我们演示出在某次执行中出现失败的情况，而在重新执行的时候，会出现成功的情况，看结果：

可以看到，用例 02 重新执行了一次就成功了，剩余的两次执行就终止了。

一条一条用例的执行，肯定会很慢，来看如何并发的执行测试用例，当然这需要相应的插件。

下载

使用

在配置文件中添加：

就是这个 -n=auto ：

并发的配置可以写在配置文件中，然后其他正常的执行用例脚本即可。另外一种就是在终端中指定，先来看示例：

结果：

pytest-sugar 改变了 pytest 的默认外观，添加了一个进度条，并立即显示失败的测试。它不需要配置，只需 下载插件即可，用 pytest 运行测试，来享受更漂亮、更有用的输出。

下载

其他照旧执行用例即可。

pytest-cov 在 pytest 中增加了覆盖率支持，来显示哪些代码行已经测试过，哪些还没有。它还将包括项目的测试覆盖率。

下载

使用

在配置文件中：

也就是配置 --cov=./scripts ，这样，它就会统计所有 scripts 目录下所有符合规则的脚本的测试覆盖率。

执行的话，就照常执行就行。

结果：

更多插件参考：https://zhuanlan.hu.com/p/50317866

有的时候，在 pytest.ini 中配置了 pytest-html 和 allure 插件之后，执行后报错：

出现了这个报错，检查你配置的解释器中是否存在 pytest-html 和 allure-pytest 这两个模块。如果是使用的pycharm ide，那么你除了检查settings中的解释器配置之外，还需要保证运行脚本的编辑器配置是否跟settings中配置一致。

⑹ 如何用python写脚本

以Python2.7操作为例：
1、首先需要打开电脑桌面，按开始的快捷键，点击Python2.7如图所示的选项进入。
相关推荐：《Python入门教程》
2、打开之后，开始编辑脚本，脚本第一行一定要写上 #!usr/bin/python表示该脚本文件是可执行python脚本，如果python目录不在usr/bin目录下，则替换成当前python执行程序的目录。
3、脚本写完之后，打开CMD命令行，开始调试、可以直接用editplus调试。
4、最后，CMD命令行中，输入 “python” + “空格”，即 ”python “，然后敲回车运行即可，这样就可以把编辑好的脚本运行了。

⑺ python写的selenium测试脚本，run.py文件引测试脚本怎么批量引入

获取以test开头,以.py结尾的测试用例create_suite

生产测试报告eport_design

#coding=utf-8

fromemail.headerimportHeader

fromemail.mime.textimportMIMEText

importsmtplib

importunittest

importtime

importsys

#reload(sys)

#sys.setdefaultencoding('utf-8')

defcreate_suite():

#1.获取框架中脚本的位置

script_dir="..Script\add"

#2.获取要运行的脚本--discover

discv=unittest.defaultTestLoader.discover(script_dir,pattern="test_add_*.py")

#3.讲获取的脚本加入到测试集合

#创建一个测试集合

suite=unittest.TestSuite()

#循环遍历discv列表中脚本的名字,并加入到suite中

forcaseindiscv:

#printcase

suite.addTest(case)

#讲测试集返回

returnsuite

defreport_design():

globalfilename,runner,file1

now=time.strftime("%Y-%m-%d%H-%M-%S")

filename=".\Ggpt\add\"+now+"result.html"

file1=open(filename,'wb+')#wb+二进制写入方式

#stream报告文件title标题description

runner=HTMLTestRunner(stream=file1,title="selenium_test_report",description="用例执行情况")

⑻ 集中式日志分析平台 - ELK Stack - Filebeat 压测

任何一款采集 agent 进行公司内全面推广前都需要进行性能测试以及资源限制功能测试，以保证：

对于 Filebeat 这款号称 golang 编写，性能强于 logstahs-forwarder 的采集 agent，我们也需要这样进行严谨对待。

硬件选择虚拟机，6cores + 16GB Mem + 175GB SSD + 1000Mbps 带宽；

Filebeat 配置，输出到 console：

Filebeat 配置，输出到 Kafka：

我们开启 Filebeat 的 6060 端口，并使用 python 脚本进行指标采集。
expvar_rates.py ，每秒统计出 Filebeat 指标，主要看：

Step1. 启动 Filebeat (172.16.134.8)

Step2. 启动统计脚本

Step3. 启动 tsar

Step4. 写入压测数据(6个进程写入，6千万条日志)

在 6 进程数据写入日志文件时，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 0.8 cores。
在 6 进程数据写入日志文件后，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 1.6 cores。
小结：

测试步骤和上述一致，区别在于配置文件需要输出到 Kafka。
在 6 进程数据写入日志文件时，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 0.7~0.8 cores。
在 6 进程数据写入日志文件后，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 2.0 cores。
小结：

测试步骤和上述一致，区别在于配置文件需要输出到 Kafka。
和上述步骤不同的是，启动 Filebeat 时需要 systemd 限制 CPU、句柄数，根据之前的理论，句柄数限制在 100 已经非常够用，CPU 限制在 1 core。
修改 /usr/lib/systemd/system/filebeat.service ：

执行 reload：

对 CPU 进行限制：

确认是否限制成功：

有如下输出表示OK：

在 6 进程数据写入日志文件时，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 0.7 ~ 0.8 cores。
在 6 进程数据写入日志文件后，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 1.0 cores，限制生效。
小结：

在 6 进程数据写入日志文件时，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 0.75 ~ 0.9 cores。
在 6 进程数据写入日志文件后，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 1.0 cores，限制生效。
小结：

在 6 进程数据写入日志文件时，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 0.7 ~ 0.75 cores。
在 6 进程数据写入日志文件后，我们在开启 python 统计脚本的窗口得到如下稳定的统计数据：

我们在 tsar 看到的统计数据为：

我们在 top 中可以看到 Filebeat 大致占据了 0.9 cores，未达到限制。
小结：

A. FB 全力采集 247B 数据(真实环境类似日志长度)，速率为 ~ 40K/s，CPU 开销为 2 cores；

B. FB 在 CPU 限制 1 cores 情况下，采集 247B 数据速率为 ~ 20K/s，可以认为单核采集速率为 ~ 20K/s/core；

C. 日志单行数据越大，吞吐越小，5KB 每行已经非常夸张，即使如此，没有压缩的情况下带宽消耗 35MBps，gzip 压缩率一般为 0.3~0.4，占用带宽为 10.5~14MBps，对于千兆网卡来说压力较小；

⑼ Python 如何写脚本

以Python2.7操作为例：

1、首先需要打开电脑桌面，按开始的快捷键，点击Python2.7如图所示的选项进入。

⑽ 初学Python，想做手机自动化测试脚本，想了解几个问题

1、手机自动化测试Python能独立完成吗？可以。
2、想要学的话，看哪本教程会好些？首先学习自动化测试，然后学习python，然后结合实例学习。可以参考http://wenku..com/view/fd8b690b581b6bd97f19ea61.html
3、主要要学习的模块内容或者方向是哪些？
Python的世界有一个开源框架Splinter，可以非常棒的模拟浏览器的行为（从某种意义上也可以说是人的访问点击行为）。Splinter提供了丰富的API，可以获取页面的信息，以判断当前的行为所产生的结果

4、还有懂这行补充给我的，我另加分。。多项
多学习测试的各方面知识，python只是工具。测试的理论知识很重要。