Locust 性能框架

🍰 Locust是一款由Python开发的开源性能测试框架，支持分布式，跨平台，易扩展，可在多台主机上对系统持续发送请求。

1 安装

  • 安装
    • 与JMeter、LoadRunner的等压测工具使用线程的方式模拟用户请求不同。
    • Locust是使用协程的方式模拟用户请求，协程的上下文切换是由自己控制。
    • 当一协程执行完成后会主动让出，让另一协程开始执行，切换是在用户态完成的。
    • 而线程切换是受系统控制，在用户态与内核态之间切换。
    • 所以协程上下文切换的代价远比线程切换的代价小的多。
    • 因此Locust可以达到更高数量级的并发，官网：https://docs.locust.io/en/stable/。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 需Python3.6以上版本支持
pip install locust

# 打算在多个进程或机器上分布式运行Locust
pip install pyzmq

# 查看Locust版本
locust -V
locust --version

# 查看Locust使用帮助
locust -h

# 启动Locust服务
locust -f file_name.py

2 使用

  • 使用
    • 定义任务集：一个类，必须继承TaskSet，类里面是用户定义的任务。
    • 定义用户类：一个类，必须继承HttpUser，配置host地址，复写tasks。
    • 定义任务(接口请求)：普通函数，必须有一个形参，使用task修饰函数。
    • 启动服务，配置执行：locust -f 文件名。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
from locust import HttpUser, TaskSet, task


# 定义任务集
class TaskTest(TaskSet):
    # 定义任务
    @task
    def say(params):
        print("正在说话")

    # 修饰函数，将函数定义成任务
    # 数字代表权重，权重越大执行次数越多，默认值1
    @task(5)
    def sing(params):
        print("正在唱歌")

    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("初始操作")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("收尾操作")


# 定义用户类
class Test(HttpUser):
    tasks = [TaskTest]
    host = "http://localhost"
    
    # 用户执行任务期间等待时间的下界，默认1000，单位毫秒
    min_wait = 1000
    
    # 用户执行任务期间等待时间的上界，默认1000，单位毫秒
    max_wait = 2000
    
    # 被选中的概率，权重越大，被选中的机会就越大，默认10
    weight = 10

2-1 单接口

  • 单接口
    • TaskSet类和SequentialTaskSet类实现了虚拟用户所执行任务的调度算法。
    • HttpUser类有一个用于发送http请求的client属性，发送请求前需先调用类。
    • task装饰器是为了给用户添加任务，只有带task装饰器的方法才会发起请求。
    • 禅道下载：https://dl.zentao.net/zentao/18.0/ZenTaoPMS.18.0.win64.exe。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
from locust import HttpUser, TaskSet, task, between, constant

# -- 清空表数据，主键id重置自增值
# -- TRUNCATE TABLE zt_todo;

# -- 获取当前自增值
# -- SHOW TABLE STATUS LIKE "zt_todo";

# -- 将自增值重置为小于等于当前已用ID的最大值，通常设为0
# -- ALTER TABLE `zt_todo` AUTO_INCREMENT = 0;


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(TaskSet):
    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task
    def login(self):
        cookies = {
            "zentaosid": "54d1af94725410432656cb753068b800",
            "lang": "zh-cn", "device": "desktop",
            "theme": "default", "hidePatch": "true"
        }

        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=cookies, catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"

    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)

    # 请求间隔时间，每次请求间隔固定3s
    # wait_time = constant(3)

2-2 多接口

  • 多接口
    • 多接口：比单接口测试复杂，关注多个接口的整体性能表现，接口之间通常无明显的依赖关系。
    • 业务流：最复杂，能真实反映用户行为，接口之间存在依赖关系，关注整个业务流的性能表现。
    • with结构并且启用catch_response=True，会影响任务权重机制，错误估计任务的执行时间。
    • 解决方法
      • 如果响应处理逻辑复杂，或需要精确测量每个请求的执行时间，则手动计时。
        • 手动计时：可在with语句块的开始和结束时，手动记录时间。
        • 并使用events.request.fire事件通知任务的实际执行时间。
      • 如果响应处理逻辑简单，并且不需要精确计时，则使用函数并手动处理响应。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
import datetime
from locust import HttpUser, TaskSet, task, between


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(TaskSet):
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)
    
    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")
    
    # 定义任务：添加待办
    @task(3)
    def create_todo(self):
        ZentaoUser.counter += 1
        todo_url = "/zentao/todo-create.html?onlybody=yes"
        
        # 获取当前时间和下一小时的时间
        now = datetime.datetime.now()
        next_hour = now + datetime.timedelta(hours=1)
        
        # 格式化时间字符串10:00
        current_time_str = now.strftime("%H:%M")
        next_hour_time_str = next_hour.strftime("%H:%M")
        current_date_str = now.strftime("%Y-%m-%d")
        
        # 定义请求头
        todo_headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": "http://127.0.0.1/zentao/todo-create.html?onlybody=yes",
        }

        # 定义POST请求的body数据
        todo_data = {
            "date": f"{current_date_str}",
            "config[day]": "",
            "config[specify][month]": "0",
            "config[specify][day]": "1",
            "config[type]": "day",
            "config[beforeDays]": "0",
            "config[end]": "",
            "type": "custom",
            "assignedTo": "admin",
            "name": "",
            "name": f"测试{ZentaoUser.counter}",
            "pri": "3",
            "desc": f"待办测试{ZentaoUser.counter}",
            "status": "wait",
            "begin": f"{current_time_str}",
            "end": f"{next_hour_time_str}"
        }

        # 发送POST请求，with影响任务权重机制，会错误地估计任务的执行时间
        # with self.client.post(
        #     cookies=ZentaoUser.cookies, headers=todo_headers,
        #     url=todo_url, data=todo_data, name="添加待办",
            
        #     # 允许自定义处理服务器的响应
        #     catch_response=True
        # ) as response:
        #     if response.status_code == 200:
        #         if "success" in response.text:
        #             response.success()
        #         else:
        #             response.failure(f"添加失败：{response.status_code}")
        #     else:
        #         response.failure(f"添加失败：{response.status_code}")
        
        # Locust会在请求完成后立即结束计时，不会考虑后续对响应的处理
        # 任务的执行时间能够更准确地反映实际的请求时间
        # Locust也能更准确地根据权重分配任务的执行频率
        self.client.request(
            method="POST", cookies=ZentaoUser.cookies,
            headers=todo_headers, url=todo_url,
            data=todo_data, name="添加待办"
        )


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"
    counter = 0
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }

(1) 手动计时

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
import time
import datetime
from locust import HttpUser, TaskSet, task, between, events


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(TaskSet):
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)
    
    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")
    
    # 定义任务：添加待办
    @task(3)
    def create_todo(self):
        ZentaoUser.counter += 1
        start_time = time.time()
        todo_url = "/zentao/todo-create.html?onlybody=yes"
        
        # 获取当前时间和下一小时的时间
        now = datetime.datetime.now()
        next_hour = now + datetime.timedelta(hours=1)
        
        # 格式化时间字符串10:00
        current_time_str = now.strftime("%H:%M")
        next_hour_time_str = next_hour.strftime("%H:%M")
        current_date_str = now.strftime("%Y-%m-%d")
        
        # 定义请求头
        todo_headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": "http://127.0.0.1/zentao/todo-create.html?onlybody=yes",
        }

        # 定义POST请求的body数据
        todo_data = {
            "date": f"{current_date_str}",
            "config[day]": "",
            "config[specify][month]": "0",
            "config[specify][day]": "1",
            "config[type]": "day",
            "config[beforeDays]": "0",
            "config[end]": "",
            "type": "custom",
            "assignedTo": "admin",
            "name": "",
            "name": f"测试{ZentaoUser.counter}",
            "pri": "3",
            "desc": f"待办测试{ZentaoUser.counter}",
            "status": "wait",
            "begin": f"{current_time_str}",
            "end": f"{next_hour_time_str}"
        }

        # 发送POST请求，with影响任务权重机制，会错误地估计任务的执行时间
        with self.client.post(
            cookies=ZentaoUser.cookies, headers=todo_headers,
            url=todo_url, data=todo_data, name="添加待办",
            
            # 允许自定义处理服务器的响应
            catch_response=True
        ) as response:
            response_time = int((time.time() - start_time) * 1000)
            if response.status_code == 200:
                if "success" in response.text:
                    response.success()
                    
                    # 通知Locust任务的实际执行时间
                    events.request.fire(
                        request_type="POST",
                        name="添加待办",
                        response_time=response_time,
                        response_length=len(response.content),
                        response=response,
                        exception=None
                    )
                else:
                    response.failure(f"添加失败：{response.status_code}")
                    events.request.fire(
                        request_type="POST",
                        name="添加待办",
                        response_time=response_time,
                        response_length=len(response.content),
                        response=response,
                        exception=f"添加失败：{response.status_code}",
                    )
            else:
                response.failure(f"添加失败：{response.status_code}")
                events.request.fire(
                    request_type="POST",
                    name="添加待办",
                    response_time=response_time,
                    response_length=len(response.content),
                    response=response,
                    exception=f"添加失败：{response.status_code}",
                )


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"
    counter = 0
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }

(2) 使用函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import datetime
from locust import HttpUser, TaskSet, task, between


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(TaskSet):
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)

    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")

    # 定义任务：添加待办
    @task(3)
    def create_todo(self):
        ZentaoUser.counter += 1
        todo_url = "/zentao/todo-create.html?onlybody=yes"

        # 获取当前时间和下一小时的时间
        now = datetime.datetime.now()
        next_hour = now + datetime.timedelta(hours=1)

        # 格式化时间字符串10:00
        current_time_str = now.strftime("%H:%M")
        next_hour_time_str = next_hour.strftime("%H:%M")
        current_date_str = now.strftime("%Y-%m-%d")

        # 定义请求头
        todo_headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": "http://127.0.0.1/zentao/todo-create.html?onlybody=yes",
        }

        # 定义POST请求的body数据
        todo_data = {
            "date": f"{current_date_str}",
            "config[day]": "",
            "config[specify][month]": "0",
            "config[specify][day]": "1",
            "config[type]": "day",
            "config[beforeDays]": "0",
            "config[end]": "",
            "type": "custom",
            "assignedTo": "admin",
            "name": "",
            "name": f"测试{ZentaoUser.counter}",
            "pri": "3",
            "desc": f"待办测试{ZentaoUser.counter}",
            "status": "wait",
            "begin": f"{current_time_str}",
            "end": f"{next_hour_time_str}"
        }

        # Locust会在请求完成后立即结束计时，不会考虑后续对响应的处理
        # 任务的执行时间能够更准确地反映实际的请求时间
        # Locust也能更准确地根据权重分配任务的执行频率
        response = self.client.request(
            method="POST", cookies=ZentaoUser.cookies,
            headers=todo_headers, url=todo_url,
            data=todo_data, name="添加待办"
        )

        # 使用self.client.request()并手动处理响应
        if response.status_code == 200:
            # 打印响应内容用于调试
            # print(response.text)

            # 检查响应是否包含重定向
            if "<script>if" in response.text:
                # 使用success需要调用catch_response=True，否则报错
                # response.success()
                print(f"添加成功：{response.text}")
            else:
                response.failure(f"添加失败：{response.status_code}")
                print(f"添加失败：{response.status_code}")
        else:
            response.failure(f"添加失败：{response.status_code}")
            print(f"添加失败：{response.status_code}")


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"
    counter = 0
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }

2-3 业务流

  • 业务流
    • 对于业务流的测试，需要使用SequentialTaskSet类。
    • 可以按代码顺序由上到下执行带有task装饰器的任务。
    • 使用TaskSet类，结合on_start()和on_stop()方法的行为将更加灵活。
    • SequentialTaskSet类则影响on_start()和on_stop()的执行方式和时机。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
import datetime
from locust import HttpUser, SequentialTaskSet, task, between


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(SequentialTaskSet):
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)

    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")

    # 定义任务：添加待办
    @task
    def create_todo(self):
        ZentaoUser.counter += 1
        todo_url = "/zentao/todo-create.html?onlybody=yes"

        # 获取当前时间和下一小时的时间
        now = datetime.datetime.now()
        next_hour = now + datetime.timedelta(hours=1)

        # 格式化时间字符串10:00
        current_time_str = now.strftime("%H:%M")
        next_hour_time_str = next_hour.strftime("%H:%M")
        current_date_str = now.strftime("%Y-%m-%d")

        # 定义请求头
        todo_headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": "http://127.0.0.1/zentao/todo-create.html?onlybody=yes",
        }

        # 定义POST请求的body数据
        todo_data = {
            "date": f"{current_date_str}",
            "config[day]": "",
            "config[specify][month]": "0",
            "config[specify][day]": "1",
            "config[type]": "day",
            "config[beforeDays]": "0",
            "config[end]": "",
            "type": "custom",
            "assignedTo": "admin",
            "name": "",
            "name": f"测试{ZentaoUser.counter}",
            "pri": "3",
            "desc": f"待办测试{ZentaoUser.counter}",
            "status": "wait",
            "begin": f"{current_time_str}",
            "end": f"{next_hour_time_str}"
        }

        # Locust会在请求完成后立即结束计时，不会考虑后续对响应的处理
        # 任务的执行时间能够更准确地反映实际的请求时间
        # Locust也能更准确地根据权重分配任务的执行频率
        response = self.client.request(
            method="POST", cookies=ZentaoUser.cookies,
            headers=todo_headers, url=todo_url,
            data=todo_data, name="添加待办"
        )

        # 使用self.client.request()并手动处理响应
        if response.status_code == 200:
            # 打印响应内容用于调试
            # print(response.text)

            # 检查响应是否包含重定向
            if "<script>if" in response.text:
                # 使用success需要调用catch_response=True，否则报错
                # response.success()
                print(f"添加成功：{response.text}")
            else:
                response.failure(f"添加失败：{response.status_code}")
                print(f"添加失败：{response.status_code}")
        else:
            response.failure(f"添加失败：{response.status_code}")
            print(f"添加失败：{response.status_code}")


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"
    counter = 0
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }

2-4 用户类

  • 用户类
    • 也可只调用HttpUser类(用户类)，通常代码量少时使用。
    • tag装饰器用于标记任务，可以执行指定带有标记的任务。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from locust import HttpUser, task, tag, between


# 定义用户类：以本地搭建的禅道为例
class ZentaoUser(HttpUser):
    host = "http://127.0.0.1"
    
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }
    
    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @tag("login")
    @task
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }
        
        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")

3 断言

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import re
from bs4 import BeautifulSoup
from locust import HttpUser, task, between


# 定义用户类：以本地搭建的禅道为例
class ZentaoUser(HttpUser):
    host = "http://127.0.0.1"
    
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }
    
    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            print(f"响应内容：{response.text}")
            
            # 解析HTML响应内容
            soup = BeautifulSoup(response.text, "html.parser")
            
            # 通过响应内容断言
            if soup.find("meta") or soup.find("script") or \
            re.search(
                r"""self\.location="/zentao/";""",
                response.text
            ):
                print("登录成功")
            else:
                print("登录失败")
            
            # 通过状态码断言
            if response.status_code == 200:
                print(f"登录成功：{response.status_code}")
            else:
                print(f"登录失败：{response.status_code}")
            
            # 通过响应时间断言
            response_time = response.elapsed.total_seconds()
            print(f"响应时间：{response_time}")
            
            if response_time < 0.3:
                print("响应时间：小于0.3秒")
            elif response_time == 1:
                print("响应时间：等于0.3秒")
            else:
                print("响应时间：大于0.3秒")

4 参数化

  • 参数化
    • 借助第三方库mimesis或faker可以进行参数化操作。
    • 自动生成账号、密码、邮箱、文本、地址、日期等。
1
2
3
4
5
6
7
# 参数化设置需借助第三方库

# https://faker.readthedocs.io/en/master/
pip install faker

# https://mimesis.name/master/
pip install mimesis

4-1 faker

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
import datetime
from faker import Faker
from locust import HttpUser, TaskSet, task, between


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(TaskSet):
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)

    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task(1)
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")

    # 定义任务：添加待办
    @task(9)
    def create_todo(self):
        todo_url = "/zentao/todo-create.html?onlybody=yes"

        # 获取当前时间和下一小时的时间
        now = datetime.datetime.now()
        next_hour = now + datetime.timedelta(hours=1)

        # 格式化时间字符串10:00
        current_time_str = now.strftime("%H:%M")
        next_hour_time_str = next_hour.strftime("%H:%M")
        current_date_str = now.strftime("%Y-%m-%d")

        # 定义请求头
        todo_headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": "http://127.0.0.1/zentao/todo-create.html?onlybody=yes",
        }
        
        # 创建一个中文环境的Faker对象
        fake = Faker("ZH_CN")

        # [:-1]去掉句子末尾的句号
        title = fake.sentence(nb_words=6)[:-1]
        description = fake.sentence(nb_words=10)

        # 定义POST请求的body数据
        todo_data = {
            "date": f"{current_date_str}",
            "config[day]": "",
            "config[specify][month]": "0",
            "config[specify][day]": "1",
            "config[type]": "day",
            "config[beforeDays]": "0",
            "config[end]": "",
            "type": "custom",
            "assignedTo": "admin",
            "name": "",
            "name": {title},
            "pri": "3",
            "desc": {description},
            "status": "wait",
            "begin": f"{current_time_str}",
            "end": f"{next_hour_time_str}"
        }

        # Locust会在请求完成后立即结束计时，不会考虑后续对响应的处理
        # 任务的执行时间能够更准确地反映实际的请求时间
        # Locust也能更准确地根据权重分配任务的执行频率
        response = self.client.request(
            method="POST", cookies=ZentaoUser.cookies,
            headers=todo_headers, url=todo_url,
            data=todo_data, name="添加待办"
        )

        # 使用self.client.request()并手动处理响应
        if response.status_code == 200:
            # 打印响应内容用于调试
            # print(response.text)

            # 检查响应是否包含重定向
            if "<script>if" in response.text:
                # 使用success需要调用catch_response=True，否则报错
                # response.success()
                print(f"添加成功：{response.text}")
            else:
                response.failure(f"添加失败：{response.status_code}")
                print(f"添加失败：{response.status_code}")
        else:
            response.failure(f"添加失败：{response.status_code}")
            print(f"添加失败：{response.status_code}")


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }

4-2 mimesis

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
import datetime
from mimesis import Text
from mimesis.locales import Locale
from locust import HttpUser, TaskSet, task, between


# 定义任务集：以本地搭建的禅道为例
class ZentaoTask(TaskSet):
    # 请求间隔时间，每次请求间隔范围在1到3s
    wait_time = between(1, 3)

    # 前置方法，在所有任务前调用一次
    def on_start(self):
        print("***开始执行测试***")

    # 后置方法，任务集停止时调用一次
    def on_stop(self):
        print("***执行测试结束***")

    # 定义任务：登录系统
    @task(1)
    def login(self):
        # 登录参数
        data = {
            "username": "admin",
            "password": "Aa.123456"
        }

        # 发送登录请求
        with self.client.post(
            "/zentao/user-login.html", data=data,
            cookies=ZentaoUser.cookies,
            catch_response=True, name="系统登录"
        ) as response:
            if response.status_code == 200:
                response.success()
            else:
                response.failure(f"登陆失败：{response.status_code}")

    # 定义任务：添加待办
    @task(9)
    def create_todo(self):
        todo_url = "/zentao/todo-create.html?onlybody=yes"

        # 获取当前时间和下一小时的时间
        now = datetime.datetime.now()
        next_hour = now + datetime.timedelta(hours=1)

        # 格式化时间字符串10:00
        current_time_str = now.strftime("%H:%M")
        next_hour_time_str = next_hour.strftime("%H:%M")
        current_date_str = now.strftime("%Y-%m-%d")

        # 定义请求头
        todo_headers = {
            "Content-Type": "application/x-www-form-urlencoded",
            "Referer": "http://127.0.0.1/zentao/todo-create.html?onlybody=yes",
        }
        
        # 使用mimesis造数据
        text = Text(locale=Locale.ZH)
        title = text.title()
        description = text.sentence()

        # 定义POST请求的body数据
        todo_data = {
            "date": f"{current_date_str}",
            "config[day]": "",
            "config[specify][month]": "0",
            "config[specify][day]": "1",
            "config[type]": "day",
            "config[beforeDays]": "0",
            "config[end]": "",
            "type": "custom",
            "assignedTo": "admin",
            "name": "",
            "name": {title},
            "pri": "3",
            "desc": {description},
            "status": "wait",
            "begin": f"{current_time_str}",
            "end": f"{next_hour_time_str}"
        }

        # Locust会在请求完成后立即结束计时，不会考虑后续对响应的处理
        # 任务的执行时间能够更准确地反映实际的请求时间
        # Locust也能更准确地根据权重分配任务的执行频率
        response = self.client.request(
            method="POST", cookies=ZentaoUser.cookies,
            headers=todo_headers, url=todo_url,
            data=todo_data, name="添加待办"
        )

        # 使用self.client.request()并手动处理响应
        if response.status_code == 200:
            # 打印响应内容用于调试
            # print(response.text)

            # 检查响应是否包含重定向
            if "<script>if" in response.text:
                # 使用success需要调用catch_response=True，否则报错
                # response.success()
                print(f"添加成功：{response.text}")
            else:
                response.failure(f"添加失败：{response.status_code}")
                print(f"添加失败：{response.status_code}")
        else:
            response.failure(f"添加失败：{response.status_code}")
            print(f"添加失败：{response.status_code}")


# 定义用户类
class ZentaoUser(HttpUser):
    tasks = [ZentaoTask]
    host = "http://127.0.0.1"
    
    # 全局共享cookies
    cookies = {
        "zentaosid": "54d1af94725410432656cb753068b800",
        "lang": "zh-cn", "device": "desktop",
        "theme": "default", "hidePatch": "true"
    }

5 UI界面

  • UI界面
    • 启动页(Start new load test)。
      • Number of users (peak concurrency)：必填，并发用户数，即虚拟用户数。
      • Ramp up (users started/second)
        • 必填，在开始进行负载测试时，用户逐渐增加到目标并发用户数的速率。
        • 若设置10用户/秒，且目标是100个用户，则需10秒才能达到最大并发量。
      • Host：被测试的目标服务器或应用程序的地址，代码中如果设置则默认执行。
      • Advanced options：允许进一步自定义测试配置，比如设置请求的超时时间等。
    • 界面模块
      • STATISTICS：统计信息，类似于JMeter的聚合报告。
      • CHARTS：图表，包括请求数量、响应时间、失败请求数等随时间变化的曲线图。
      • FAILURES：失败情况，包括失效的请求类型、URL、失败原因及响应状态码等。
      • EXCEPTIONS：异常情况，包括代码错误、系统异常等。
      • CURRENT RATIO：当前成功请求与失败请求的比率，百分比形式呈现。
      • DOWNLOAD DATA：测试数据的下载模块，提供了四种类型数据的下载。
      • LOGS：日志，包括请求的详细记录、异常信息以及其他相关的调试信息。
      • LOCUST CLOUD：Locust云服务，提供关于云环境中运行的测试的信息。

5-1 STATISTICS

  • STATISTICS
    • Type：类型，请求的类型，POST、GET、INPUT等。
    • Name：名称，当前进行负载测试的目标URL或服务名称。
    • Requests：请求数，在负载测试期间，发送的总请求数量。
    • Fails：失败数，测试过程中发生的请求失败数量，包括因超时等导致的请求失效情况。
    • Median (ms)：中位数响应时间，毫秒，表示一半请求的响应时间低于此值。
    • 95%ile (ms)：95%百分位响应时间，毫秒，可帮助识别大多数用户的响应速度。
    • 99%ile (ms)：99%百分位响应时间，毫秒，发现极端慢的请求，帮助分析瓶颈。
    • Average (ms)：平均响应时间，毫秒，所有请求的响应时间平均值，反映整体性能表现。
    • Min (ms)：最小响应时间，毫秒，显示最快的请求响应时间。
    • Max (ms)：最大响应时间，毫秒，显示最慢的请求响应时间。
    • Average size (bytes)：平均响应大小，字节，帮助评估带宽使用情况。
    • Current RPS：当前每秒请求数，反映实时的处理能力。
    • Current Failures/s：当前每秒失败数，监控系统的健康状况，及时发现问题。
    • Aggregated：聚合统计，显示在指定时间范围内的汇总数据，展示整体表现。

5-2 CHARTS

  • CHARTS
    • Total Requests per Second (RPS)
      • 该图表展示在负载测试过程中，每秒钟发送的请求总数。
      • 反映了系统在不同时间段内处理请求的能力和负载情况。
      • 通过观察曲线了解请求量的变化趋势，从而识别高峰期或潜在的性能瓶颈。
    • Response Times (ms)
      • 该图表展示请求的响应时间分布，包括中位数、95%百分位、最大响应时间等。
      • 通过查看响应时间的变化，可评估应用程序的性能以及在不同负载下的稳定性。
    • Number of Users
      • 该图表展示了在执行负载测试期间，活跃的虚拟用户数量。
      • 随测试的进行，用户数量变化反映出系统的并发处理能力。

6 文件执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
import datetime
from locust import HttpUser, SequentialTaskSet, task, tag, between


# 定义任务集
class TaskTest(SequentialTaskSet):
    # 定义任务，多个标签
    @tag("say", "people")
    @task
    def say(params):
        print("正在说话")

    @tag("eat", "action")
    @task
    def eat(params):
        print("正在吃饭")

    @tag("sing", "people")
    @task
    def sing(params):
        print("正在唱歌")

    @tag("dance", "people")
    def dance(params):
        print("正在跳舞")

    def sleep(params):
        print("正在睡觉")

    def on_start(self):
        start_time = datetime.datetime.now()
        print("***初始操作***")
        print(f"开始时间：{start_time}")

    def on_stop(self):
        end_time = datetime.datetime.now()
        print(f"结束时间：{end_time}")
        print("***收尾操作***")


# 定义用户类
class Test(HttpUser):
    tasks = [TaskTest]
    host = "http://localhost"
    wait_time = between(1, 3)

6-1 GUI模式

1
2
3
4
5
6
7
8
9
10
11
# 执行文件中所有带@task装饰器的任务
locust -f file_name.py

# 执行文件中标签为login、info的任务
locust -f file_name.py --tags login info

# -f：指定一个Python文件
# --host：指定被测试的目标主机
# --run-time：指定测试运行的时间(时分秒：h、m、s)
# --stop-timeout：指定停止测试后等待所有用户完成当前任务的时间
locust -f file_name.py --host=http://www.xxx.com --run-time=20s --stop-timeout=10s

6-2 非GUI模式

1
2
3
4
5
6
7
# --headless：必需，指定非UI模式运行，与--no-web类似
# -u：必需，即--users，指定虚拟用户的数量(目标用户数)
# -r：必需，指定每秒启动的用户数
# -t：必需，指定测试持续的时间，小时、分钟、秒(h、m、s)
# --html：可选，生成HTML测试报告
# --csv：可选，生成CSV测试报告
locust -f file_name.py --headless -u 10 -r 2 -t 50s --html=./report/report_name.html

6-3 分布式压测

  • 分布式压测
    • 每台测试机上都要安装Locust，并且需将测试脚本拷贝到所用的每台测试机上。
    • 如果主机和从机在同一台机器上，--master-host和--master-port可以不写。
    • 但主从机共享同一台机器的计算资源，高负载下容易导致性能瓶颈，不推荐使用。
    • 通常情况下，主机节点不用参与实际压测，主要职责是协调从机节点和汇总数据。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
# 在主机上执行命令，GUI界面会在主机上运行，从机主要任务是执行测试
locust -f file_name.py --master

# 在每台从机上执行命令，指定主机IP
locust -f file_name.py --worker --master-host=192.168.xxx.xxx

# --expect-workers：等待从机连接再开始测试，从机个数达到要求后会立即开始执行
locust -f file_name.py --master --expect-workers=2 --headless -u 10 -r 2 -t 5s

# 在每台从机上执行命令，指定主机IP
locust -f file_name.py --worker --master-host=192.168.xxx.xxx --headless

# --master-bind-host：指定主节点绑定的地址，默认0.0.0.0，允许从任何地方访问
# --master-bind-port：指定主节点绑定的端口，默认5557
locust -f file_name.py --master --master-bind-host=0.0.0.0 --master-bind-port=5557

7 性能指标

  • 性能指标
    • 压力测试：强负载下的测试(大数据量、大量并发用户等)。
    • 压测分为
      • 极限负载情况下导致系统崩溃的破坏性压力测试。
      • 高负载下长时间(例如24小时)的稳定性压力测试。
    • 负载测试：不断加载方式观察不同负载下系统的响应时间、吞吐量、资源利用率等。
    • 基本性能指标：响应时间、吞吐量、资源利用率，使用Locust自带的监控功能即可。
    • 性能测试首要关注的是CPU，通常一个系统会随着压力的增大而升高CPU的利用率。
    • 单台测试机的CPU处理能力有限，所以通常采用分布式压测。
    • 常见的系统瓶颈包括线程阻塞、数据库加锁、磁盘IO阻塞等。
    • 性能指标
      • 在数据库中，一个事务通常包含了多个DML(增删改操作)语句。
      • PV：指页面浏览量，每次页面被加载或刷新时，PV就增加一次。
      • UV：指独立访客数，同一用户在一段时间内多次访问网站，只会被计算一次。
      • PV和UV通常一起使用来分析网站流量，PV/UV的比率可反映用户的访问深度。
        • 高比率：说明用户对网站内容感兴趣，浏览页面较多，用户粘性较好。
        • 低比率：说明用户访问深度较低，可能需要优化网站内容或用户体验。
      • RPS：指每秒内系统处理的请求数，可以是访问URL、API调用，设置数据库查询。
      • QPS：指每秒内系统处理的查询请求数，执行一次SELECT语句，则QPS就增加一次。
      • TPS：指每秒内系统处理的事务数，执行一个事务完整的DML操作，TPS就增加一次。
    • PV/UV衡量的是网站的整体流量，QPS衡量数据库的查询性能(读操作)。
    • RPS衡量服务器处理请求能力，TPS衡量数据库的事务处理能力(写操作)。
测试框架
#Python #Locust #接口 #性能
Locust 性能框架
https://stitch-top.github.io/2025/03/19/ce-shi-kuang-jia/tf07-locust/tf01-locust-xing-neng-kuang-jia/
作者
Dr.626
发布于
2025年3月19日 21:55:00
许可协议