项目实战2-测试工具开发

点击这里,观看项目说明视频讲解

我们通过一个项目来锻炼 Python 做测试开发工具的能力。

可以微信咨询 byhy44 ,购买 参考代码 和 视频讲解 。

概述

实战要求开发一款性能测试工具 黑羽压测Qt版 ,可以用来做 基于HTTP API接口 的性能测试。

要求做成一个 MDI 多功能子窗口的 图形界面程序,方便公司内部使用。 界面如下

image


本次锻炼涉及的知识点包括:

  • Qt图形界面开发的各个要点:

    菜单栏、工具栏、dockwindow、树控件、表格控件、字体图标的使用、MDI 多子窗口、控件动态边界调整、上下文菜单、编辑框文本语法高亮、动态曲线图、matplotlib作图。

  • Socket编程

    使用 UDP Socket 来接收 压测进程的统计数据,并且可视化呈现

  • 多进程外部程序调用

    启动独立的新压测进程,而不是在图形界面进程中运行压测。


下面我们通过分阶段的实战练习,来一步步的开发这个软件。

安装 hyload 库

因为要开发的 性能测试工具 要高效发送 HTTP请求对被测系统进行性能测试,这就必须要有一个高效收发HTTP请求的库。

这里使用白月黑羽自己研发的 hyload 库。

安装非常简单,直接运行 pip install hyload==0.0.2 即可。


既然是API接口 压测工具, 就需要一个API服务端 测试你的工具。

可以使用 http://httpbin.org/uuid 这个网址进行测试

实战1:主窗口界面

点击这里 观看具体的题目要求


实现一个主窗口界面,功能包括

  • 构建菜单栏、工具栏

    先实现在 菜单栏、工具栏 打开项目目录 功能。

    另外,給主窗口添加产品图标

  • 实现日志停靠窗口

    程序各模块可以调用打印日志库函数, 日志窗口显示这些打印信息

    日志窗口可以停靠在主窗口的下面、右侧,也可以单独分离出主窗口。

    设置日志窗口的最大显示行数为1000行。

    超过日志窗口可见范围,要能始终显示最新打印的内容。

  • 侧边栏和图标字体

    我们通常会将常用的操作, 放到工具栏上。

    但是有时工具栏上的按钮很多, 其中有些最常用的,往往希望用醒目的大图标单独放置一处。

    上图中,垂直左侧边栏就是一个很好的实现方式。

    其中的图标当然可以自己制作图片,但是比较麻烦。 学过前端开发的朋友知道,有图标字体,比如 font awesomeMaterial 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)
    '''
        },
    ]

点击运行按钮后,程序要做如下处理

  1. 先创建如下这些行代码,作为前缀部分
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 * ###################


  1. 分析编辑框获取的代码, 将性能场景定义中所有的 createClients 里面的客户端定义代码找到。

比如, 这样的 代码

createClients(
    'client_1', # 客户端名称
    3,       # 客户端数量
    0.1,     # 启动间隔时间,秒
    )

sleep(2)
    
createClients(
    'client_2', # 客户端名称
    3,       # 客户端数量
    0.1,     # 启动间隔时间,秒
    )

里面就涉及到两个客户端定义, client_1client_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后产生的代码后面,

后面再加上性能测试场景定义里面的代码


  1. 在步骤2产生的代码 后面添加如下代码

################## write your code * end * ###################

gevent.wait()

然后,存入压测项目根目录下的文件 run.py 中 。

再额外启动一个新的 Python进程,运行这个run.py 文件。

实战6:Socket编程 统计数据的接收

点击这里 观看具体的题目要求。

运行前面开发的性能测试run.py时,

如果提供了命令行参数 console 指定IP地址和端口,格式如下

console=127.0.0.1:18444

启动的 压力测试进程 就会将每秒的统计信息 序列化为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秒统计数据,实现动态刷新,如下图所示

image

实战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 库对这些文件中的统计数据作图,如下所示

image


其中 :

  • 第1张图显示 rps (每秒请求数量)

  • 第2张图显示 tps (每秒响应数量,蓝色)、eps(每秒错误数量,红点)、tops(每秒超时数量,绿点)

  • 第3张图显示 lps (平均每秒响应时间)

您需要高效学习,找工作? 点击咨询 报名实战班

点击查看学员就业情况

上一页