项目实战2-测试工具开发
我们通过一个项目来锻炼 Python 做测试开发工具的能力。
可以微信咨询 byhy44 ，参加实战班 或者 购买视频讲解和代码 。
您需要高效学习，找工作？ 点击咨询 报名实战班
点击查看学员就业情况
概述
实战要求开发一款性能测试工具 黑羽压测Qt版 ，可以用来做 基于HTTP API接口 的性能测试。
要求做成一个 MDI 多功能子窗口的 图形界面程序，方便公司内部使用。 界面如下
本次锻炼涉及的知识点包括：
- Qt图形界面开发的各个要点：
菜单栏、工具栏、dockwindow、树控件、表格控件、字体图标的使用、MDI 多子窗口、控件动态边界调整、上下文菜单、编辑框文本语法高亮、动态曲线图、matplotlib作图。
- Socket编程
使用 UDP Socket 来接收 压测进程的统计数据，并且可视化呈现
- 多进程外部程序调用
启动独立的新压测进程，而不是在图形界面进程中运行压测。
下面我们通过分阶段的实战练习，来一步步的开发这个软件。
安装 hyload 库
因为要开发的 性能测试工具 要高效发送 HTTP请求对被测系统进行性能测试，这就必须要有一个高效收发HTTP请求的库。
这里使用白月黑羽自己研发的 hyload 库。
安装非常简单，直接运行 pip install hyload 即可。
注意，这个hyload库的使用后续有了一些改变，和视频说明有些不同。
到了要使用这个库时，具体的使用请点击这里，参考教程里面的做法 。
既然是API接口 压测工具， 就需要一个API服务端 测试你的工具。
可以使用 http://httpbin.org/uuid 这个网址进行测试
实战1：主窗口界面
点击这里 观看具体的题目要求
实现一个主窗口界面，功能包括
- 构建菜单栏、工具栏
先实现在 菜单栏、工具栏 打开项目目录 功能。
另外，給主窗口添加产品图标
- 实现日志停靠窗口
程序各模块可以调用打印日志库函数， 日志窗口显示这些打印信息
日志窗口可以停靠在主窗口的下面、右侧，也可以单独分离出主窗口。
设置日志窗口的最大显示行数为1000行。
超过日志窗口可见范围，要能始终显示最新打印的内容。
- 侧边栏和图标字体
我们通常会将常用的操作， 放到工具栏上。
但是有时工具栏上的按钮很多， 其中有些最常用的，往往希望用醒目的大图标单独放置一处。
上图中，垂直左侧边栏就是一个很好的实现方式。
其中的图标当然可以自己制作图片，但是比较麻烦。 学过前端开发的朋友知道，有图标字体，比如 font awesome ， Material Icons ，内置了一套图标，可以直接拿来使用，非常方便。
如何在 Python Qt 程序中使用这些图标字体呢？
请大家自己搜索资料实现。
实战2：MDI 子窗口功能
点击这里 观看具体的题目要求
先学习 前面教程 常用控件4 中， MDI 多个子窗口 这一节的内容。
实现 点击侧边栏 中的 各个功能图标， 可以打开一个 MDI 子窗口。
各个子窗口中的具体功能后面再实现。
本次锻炼要做到， 多次点击同一个功能图标，比如 客户端 图标， 只会打开同一个 MDI 子窗口，而不是打开多个客户端子窗口。
实战3：API客户端1
点击这里 观看具体的题目要求
API 客户端是实现 可以编辑代码构建各种格式的HTTP请求 ，发送给被测服务端。
该功能实现后， 其实就是一个API接口测试的工具，类似 Postman这些工具。
本次练习先实现 左侧客户端代码文件管理框。
默认的客户端代码文件 放置在项目根目录下面的 client 目录中，
包括
-
显示client 目录中所有的代码文件， 为示例界面中的一个条目
-
可以对代码文件条目进行 增、删、改名
点击工具栏增加条目图标，可以添加一个代码文件
右键点击一个条目，显示
双击节点，可以改变其文件名
注意： 增、删、改名文件时， 标题栏显示内容对应的改变
- 代码目录刷新
实战4：API客户端2
点击这里 观看具体的题目要求
本次练习实现具体的 HTTP API 接口测试 代码编辑、执行功能。
当然第一反应，是应该使用 流行的 Requests 库 来 构建和 收发 HTTP消息。
但是 Requests 库不适合做性能测试， 它的性能很差，所以这里我们使用 白月黑羽自己研发的hyload库。
该库的使用详情，请参考这个文档
本次练习要实现的功能包括：
- 子窗口界面组成
包括3个部分： 左侧文件浏览器框、中间代码编辑框、右侧代码助手框
3个部分的大小可以拖到边界调整
- 代码助手功能
实现 点击 右侧代码助手框里面的条目 后， 代码自动填充。
最好是，可以根据 配置项 动态实现。
条目和对应插入代码关系如下所示
codeSnippets = [
{
'name': '创建 HTTP 客户端',
'code': '''# 创建客户端
client = HttpClient('127.0.0.1', # 目标地址:端口
timeout=10 # 超时时间，单位秒
)
'''
},
{
'name': '创建 HTTPS 客户端',
'code': '''# 创建客户端
client = HttpsClient('127.0.0.1', # 目标地址:端口
timeout=10 # 超时时间，单位秒
)
'''
},
{
'name': '使用代理',
'code': '''client.proxy('127.0.0.1:8888')
'''
},
'separator',
{
'name': '发送 简单请求',
'code': '''# 请求方法对应HTTP方法，包括：get、post、put、delete 等
response = client.get(
'/api/path1' # 请求URL
)
'''
},
{
'name': '设置 url参数',
'code': '''response = client.get(
'/api/path1',
# 通过params传入url参数
params={
'param1':'value1',
'param2':'value2'
}
)
'''
},
{
'name': '设置 消息头',
'code': '''response = client.get(
'/api/path1',
# 通过headers传入指定消息头
headers={
'header1':'value1',
'header2':'value2'
})
'''
},
{
'name': '设置 消息体，urlencode 格式',
'code': '''response = client.post(
'/api/path1',
# 通过data传入指定urlencode格式的消息体参数
data={
'param1':'value1',
'param2':'value2'
})
'''
},
{
'name': '设置 消息体，json 格式',
'code': '''response = client.post(
'/api/path1',
# 通过json传入指定json格式的消息体参数
json={
'param1':'value1',
'param2':'value2'
})
'''
},
{
'name': '设置 消息体，直接写入 字节',
'code': """response = client.post(
'/api/path1',
headers={
'Content-Type':'application/xml'
}
# 下面填写bytes的内容，注意最后的编码格式
data='''
<?xml version="1.0" encoding="UTF-8"?>
<CreateBucketConfiguration>
<StorageClass>Standard</StorageClass>
</CreateBucketConfiguration>
'''.encode('utf8')
)
"""
},
{
'name': '循环发10个请求',
'code': '''for i in range(10):
response = client.get('/api/path1')
sleep(5) # 间隔5秒
'''
},
'separator',
{
'name': '查看 响应时长',
'code': '''print(f"响应时长为 {response.responseTime} ms")
'''
},
{
'name': '查看 响应状态码',
'code': '''print(f"响应状态码为 {response.status_code} ")
'''
},
{
'name': '查看 消息体 文本内容',
'code': '''print(f"消息体字符串为 {response.text()} ")
'''
},
{
'name': '查看 消息体 原始内容',
'code': '''print(f"消息体字节串为 {response.raw} ")
'''
},
{
'name': '查看 消息体 json格式',
'code': '''pprint(response.json())
'''
},
{
'name': '查看 消息头',
'code': '''# 获取消息头Content-Type值
ct = response.headers['Content-Type']
print(f"消息头Content-Type值为 {ct} ")
'''
},
{
'name': '报告一个错误',
'code': '''
Stats.oneError() # 报告一个错误，加入到统计信息中
TestLogger.write('这里写详细信息到测试日志中，方便定位问题')
'''
},
{
'name': '测试日志添加信息',
'code': '''TestLogger.write('这里写日志信息')
'''
},
'separator',
{
'name': '等待10秒',
'code': '''sleep(10)
'''
},
]
- 点击其中一个文件条目，就加载该文件内容到代码编辑框中
标题栏显示当前编辑文件的名字
编辑过程中，点击保存按钮，可以保存编辑框内容到当前编辑文件中。
切换代码文件时，如果当前文件正在编辑，请自动保存后，再切换到新文件。
- 代码编辑框中Python代码语法高亮显示
请大家自己网上搜索相应资料实现
- 点击运行按钮，启动新进程，执行代码
注意：点击运行按钮，如果该文件没有保存，要自动保存。
然后获取编辑框的代码，并且在前面添加如下这几行代码，存入压测项目根目录的一个文件 run.py 。
然后额外启动一个新的python进程，运行这个run.py 文件。
为什么要额外启动新进程，而不是就在当前程序中运行该代码文件呢？可以自己思考一下。
import time
from pprint import pprint
from time import sleep
from hyload.httpclient import HttpsClient,HttpClient
from hyload.logger import TestLogger
from hyload.stats import Stats
# 此参数只有被性能场景调用时才会传入
arg = None
实战5：性能场景定义
点击这里 观看具体的题目要求。
黑羽压测工具的性能测试，就是先定义客户端行为，模拟单独的用户。
然后定义 性能场景， 就是 写代码 指定启动 n个 前面定义好的客户端， 从而模拟大量用户使用系统的场景。
本次练习就是实现 定义性能场景的代码。
界面组成和功能和 前面定义客户端类似， 只是代码助手的条目内容不同，最终执行时产生的代码文件和执行命令参数不同。
代码助手条目如下
cmds_zh = [
{
'name': '启动3个客户端',
'code': '''createClients(
'client_1', # 客户端名称
3, # 客户端数量
0.1, # 启动间隔时间，秒
)
'''
},
{
'name': '启动3个客户端，反复执行',
'code': '''createClientsAndKeep(
'client_1', # 客户端名称
3, # 客户端数量
0.1, # 启动间隔时间，秒
)
'''
},
{
'name': '启动3个带参数客户端',
'code': '''# 定义每个客户端对应的参数
args = ['user1','user2','user3']
createClients(
'client_1', # 客户端名称
3, # 客户端数量
1, # 启动间隔时间，秒
args # 客户端参数
)
'''
},
{
'name': '启动3个带参数客户端，反复执行',
'code': '''# 定义每个客户端对应的参数
args = ['user1','user2','user3']
createClientsAndKeep(
'client_1', # 客户端名称
3, # 客户端数量
1, # 启动间隔时间，秒
args # 客户端参数
)
'''
},
{
'name': '等待10秒',
'code': '''
sleep(10)
'''
},
]
点击运行按钮后，程序要做如下处理
- 先创建如下这些行代码，作为前缀部分
from gevent import monkey
monkey.patch_all()
from gevent import spawn
import gevent
import time
from pprint import pprint
from time import sleep
from hyload.stats import Stats
from hyload.logger import TestLogger
from hyload.httpclient import HttpsClient,HttpClient
clientName2Func = {}
# 如果 args 有值，一定是列表，元素依次赋值给每次clientfunc调用
def createClients(clientName, clientNum, interval, args=None):
clientFunc = clientName2Func[clientName]
for i in range(clientNum):
if args:
spawn(clientFunc, args[i])
else:
spawn(clientFunc)
if i < clientNum - 1:
sleep(interval)
# 如果 args 有值，一定是列表，元素依次赋值给每次clientfunc调用
def createClientsAndKeep(clientName, clientNum, interval, args=None):
clientFunc = clientName2Func[clientName]
def realFunc(args=None):
while True:
try:
clientFunc(args)
except Exception as e:
print(e)
for i in range(clientNum):
if args:
spawn(realFunc, args[i])
else:
spawn(realFunc)
if i < clientNum - 1:
sleep(interval)
Stats.start()
################## write your code * begin * ###################
- 分析编辑框获取的代码， 将性能场景定义中所有的
createClients里面的客户端定义代码找到。
比如， 这样的 代码
createClients(
'client_1', # 客户端名称
3, # 客户端数量
0.1, # 启动间隔时间，秒
)
sleep(2)
createClients(
'client_2', # 客户端名称
3, # 客户端数量
0.1, # 启动间隔时间，秒
)
里面就涉及到两个客户端定义， client_1 和 client_2 。
需要你的程序找到对应的代码文件，读入其内容后，放置到如下的函数定义中
def client_1(arg=None):
# 这里写入 client_1 的 文件内容，并且产生缩进
def client_2(arg=None):
# 这里写入 client_2 的 文件内容，并且产生缩进
# 最后再加上如下的定义关系
clientName2Func['client_1'] = client_1
clientName2Func['client_2'] = client_2
把上面的代码添加到步骤1后产生的代码后面，
后面再加上性能测试场景定义里面的代码
- 在步骤2产生的代码 后面添加如下代码
然后，存入压测项目根目录下的文件 run.py 中 。
再额外启动一个新的 Python进程，运行这个run.py 文件。
实战6：Socket编程 统计数据的接收
点击这里 观看具体的题目要求。
运行前面开发的性能测试run.py时，
如果提供了命令行参数 console 指定IP地址和端口，格式如下
启动的 压力测试进程 就会将每秒的统计信息 序列化为json格式，并且以 UDP 数据报 发送到 指定的地址和端口。
上面的示例中，就会发送到 127.0.0.1 本地地址，18444 UDP端口上。
接收到的字节串解码为字符串的内容如下所示
{"t": 1637842350, "rps": 1, "tops": 0, "eps": 0, "tps": 0, "respTimeSum": 0, "avgRespTime": 0, "total": {"send": 1, "recv": 0}}
{"t": 1637842351, "rps": 0, "tops": 0, "eps": 0, "tps": 1, "respTimeSum": 0.4268, "avgRespTime": 0.4268, "total": {"send": 1, "recv": 1, "100-500ms": 1}}
{"t": 1637842352, "rps": 1, "tops": 0, "eps": 0, "tps": 1, "respTimeSum": 0.3913, "avgRespTime": 0.3913, "total": {"send": 2, "recv": 2, "100-500ms": 2}}
{"t": 1637842353, "rps": 1, "tops": 0, "eps": 0, "tps": 1, "respTimeSum": 0.2225, "avgRespTime": 0.2225, "total": {"send": 3, "recv": 3, "100-500ms": 3}}
每行代表一次发送的消息，具体含义目前不需要知道，后续练习会有讲解
这次任务，就是 实现UDP Socket 接受 压力测试进程 的统计数据，并且滚屏显示在日志框中。
当然，要实现：在前面已经实现的启动压力测试进程的 命令行 中 加上 console=<IP>:<Port> 参数。
注意：要确保使用的端口号没有被占用，如果被占用，应该找到并使用 一个没有占用的端口号。
关于 UDP Socket 编程，可以学习这里的教程
实战7：实时监控 - 动态表格
点击这里 观看具体的题目要求。
压力测试进程 发送的统计数据是json格式，
各字段的含义如下
{
"t": 1636532998, # 时间戳，1970年距离数字格式
"rps": 24, # 该秒发送请求数量
"tps": 24, # 该秒接收响应数量
"tops": 0, # 该秒超时数量
"eps": 0, # 该秒错误数量
"respTimeSum": 5.7846, # 该秒累计响应时长，方便多个worker时统计
"avgRespTime": 0.2083 # 该秒平均响应时长
"total": {
"send": 7981, # 累计请求数量
"recv": 7965, # 累计响应数量
"timeout": 11, # 累计超时数量，如果为0 则该字段不存在
"error": 1, # 累计超时数量，如果为0 则该字段不存在
"100-500ms": 7820, # 响应时长在 100-500ms 之间的请求数量
"500-1000ms": 141, # 响应时长在 500-1000ms 之间的请求数量
"1000-3000ms": 4 # 响应时长在 1000-3000ms 之间的请求数量
}
}
具体说明，参考视频里面的讲解
本次练习要实现实时统计子窗口界面的表格部分。
表格展示 最近10秒统计数据，实现动态刷新，如下图所示
实战8：实时监控 - 曲线图
点击这里 观看具体的题目要求。
实现监控统计界面里面两种图表的作图：请求响应实时曲线图 和 响应时长曲线图
其中：请求响应实时曲线图 实时展示最近10次 每秒RPS 和 TPS 两个曲线，X轴坐标单位是 时间几分几秒，Y轴坐标单位是个
响应时长曲线图 展示最近10次 每秒平均响应时长 曲线，X轴坐标单位是 时间几分几秒，Y轴坐标单位是秒
实战9：性能统计数据matplotlib作图
点击这里 观看具体的题目要求。
如果压力进程的启动命令中有 statsfile 参数，压力进程执行过程中，就会把统计数据写入到改参数指定的文件中。
比如
c:\python38\python.exe run.py console=127.0.0.1:18444 statsfile=D:/t1/loadtest2/stats_perf/20211128-101224.sts
本次练习，需要实现统计记录数据作图功能。
-
先在性能场景脚本运行命令行中，添加 statsfile 参数，其值为 当前项目目录的 stats_perf 子目录 里面的文件。文件名为当前日期时间，扩展名为sts。
-
然后在工具栏添加一个作图按钮，点击该按钮，可以让用户选择统计数据文件（可以多选）， 然后使用 matplotlib 库对这些文件中的统计数据作图，如下所示
其中 ：
-
第1张图显示 rps （每秒请求数量）
-
第2张图显示 tps （每秒响应数量，蓝色）、eps（每秒错误数量，红点）、tops（每秒超时数量，绿点）
-
第3张图显示 lps （平均每秒响应时间）