locust入门系列篇一

梦幻小丑灯

目录
一、安装

二、常见使用方法
1、直接HttpUser中（常见，一般是代码量少）
2、写在模块中（少见）
3、单独写在taskset中（常见）

三、运行

四、TaskSet与SequentialTaskSet区别
场景一
场景二

五、代码执行顺顺序

六、参数化
场景一
场景二
场景三
一、安装

官方：https://docs.locust.io/en/stable/installation.html

1. pip3 install locust
   
2. 
   
3. # 查看版本
   
4. locust -V

复制代码

二、常见使用方法1、直接HttpUser中（常见，一般是代码量少）

1. from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
   
2. class Login(HttpUser):
   
3.     host = 'http://1.72.24.40:23'
   
4.     # wait_time = constant(3)  # 每次请求停顿时间 （思考时间）
   
5.     wait_time = between(1, 3)
   
6.     @task
   
7.     def login(self):
   
8.         url = '/login'
   
9.         json = {            
   
10.             "password": "45",
    
11.             "username": "34",
    
12.         }
    
13.         header = {"Content-Type": "application/json;charset=UTF-8"}
    
14.         self.client.request(method='POST', url=url, json=json, headers=header, name='登录1')

复制代码

2、写在模块中（少见）

1. from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
   
2. @task
   
3. def login(param):
   
4.     url = '/login'
   
5.     json = {            
   
6.         "password": "45",
   
7.         "username": "34",
   
8.     }
   
9.     header = {"Content-Type": "application/json;charset=UTF-8"}
   
10.     param.client.request(method='POST', url=url, json=json, headers=header, name='登录1')
    
11. 
    
12. @task
    
13. def home(param):
    
14.     url = '/#/login'
    
15.     param.client.request(method='GET', url=url, name='首页')
    
16. 
    
17. class Login(HttpUser):
    
18.     host = 'http://1.72.24.40:23'
    
19.     # wait_time = constant(3)  # 每次请求停顿时间 （思考时间）
    
20.     wait_time = between(1, 3)
    
21.     tasks = [login, home]

复制代码

3、单独写在taskset中（常见）

1. from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
   
2. 
   
3. 
   
4. class TaskTest(TaskSet):
   
5.     @task
   
6.     def login(self):
   
7.         url = '/login'
   
8.         json = {
   
9.             "password": "rewr",
   
10.             "username": "rewr",
    
11.         }
    
12.         header = {"Content-Type": "application/json;charset=UTF-8"}
    
13.         self.client.request(method='POST', url=url, json=json, headers=header, name='登录')
    
14. 
    
15. 
    
16. 
    
17. 
    
18. class Login(HttpUser):
    
19.     host = 'http://s2.x.x.x:x'
    
20.     # wait_time = constant(3)  # 每次请求停顿时间 （思考时间）
    
21.     wait_time = between(1, 3)
    
22.     tasks = [TaskTest]
    

复制代码

三、运行

1. locust -f 文件路径+文件名.py

复制代码

浏览器输入：http://localhost:8089/

四、TaskSet与SequentialTaskSet区别

继承TaskSet类，那么对应里面的方法是并行执行的，继承SequentialTaskSet类，那么对应里面的方法是串行执行的。

场景一

1. 例如：登录一次，添加50次，那么可以考虑使用继承SequentialTaskSet类，然后在装饰器task当中进行设置

复制代码

场景二

1. 配合on_start、on_stop（这里例子没有使用到）使用askSet然后通过配置task。on_start、on_stop是每个用户开始和结束的时候都会执行一次。
   
2. 
   
3. 模拟100个用户，先登录系统，然后90%执行A操作，10%执行B操作

复制代码

1. # -*- coding: utf-8 -*-
   
2. # @Time    : 2021/6/14 8:17
   
3. # @Project : virlocustdemo
   
4. # @Author  : testing
   
5. # @File    : locus2ss.py
   
6. # @Software: PyCharm
   
7. from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
   
8. 
   
9. 
   
10. class TaskTest(TaskSet):
    
11.     """
    
12.     这里继承的是串行执行的类
    
13.     """
    
14.     token = None
    
15. 
    
16.     # 用户执行task前调用
    
17.     def on_start(self):
    
18.         url = '/XX/XX/login'
    
19.         json = {           
    
20.             "password": "888X23338",
    
21.             "username": "XXXX"
    
22.         }
    
23.         header = {"Content-Type": "application/json}
    
24.         try:
    
25.             with self.client.post(url=url, json=json, headers=header, name='登录') as response:
    
26.                 self.token = 'Bearer' + ' ' + res['data']['access_token']
    
27.         except Exception as e:
    
28.             print("错误信息为", e)
    
29.     # 用户执行task后调用
    
30.     def on_stop(self):
    
31.         print("用户结束")
    
32.     @task(90)
    
33.     def query_users(self):
    
34.         url = '/DSAD/userMDSADe/DS'
    
35.         json = {
    
36.             "pageIndex": "1",
    
37.             "pageSize": "200"
    
38.         }
    
39.         header = {
    
40.             "Content-Type": "application/json,
    
41.             "Authorization": self.token}
    
42.         try:
    
43.             self.client.get(url=url, params=json, headers=header, name='A')
    
44.         except Exception as e:
    
45.             print("错误信息为", e)
    
46. 
    
47. 
    
48.     @task(10)
    
49.     def query_oErg_tEWee(self):
    
50.         url = '/uWEW/usEWEWe/orEWEree'
    
51.         json = {"enRdLeERvTel": "1"}
    
52.         header = {
    
53.             "Content-Type": "application/json,
    
54.             "Authorization": self.token}
    
55.         try:
    
56.             self.client.get(url=url, params=json, headers=header, name='B')
    
57.         except Exception as e:
    
58.             print("错误信息为", e)
    
59. class Login(HttpUser):
    
60.     host = 'http://X2.122.123.20:138'
    
61.     # wait_time = constant(3)  # 每次请求停顿时间 （思考时间）
    
62.     wait_time = between(1, 3)
    
63.     tasks = [TaskTest]

复制代码

五、代码执行顺顺序

测试项目执行前会执行一次SETUP，测试项目结束后会执行一次TEARDOWN。

1. import time
   
2. from locust import HttpUser,task,between,events
   
3. import urllib3
   
4. urllib3.disable_warnings()
   
5. 
   
6. @events.test_start.add_listener
   
7. def on_test_start(**kwargs):
   
8.     print('===测试最开始提示===')
   
9. 
   
10. @events.test_stop.add_listener
    
11. def on_test_stop(**kwargs):
    
12.     print('===测试结束了提示===')
    
13. 
    
14. 
    
15. class TestTask(HttpUser):
    
16. 
    
17.     wait_time = between(1, 5)
    
18.     host = 'https://www.baidu.com'
    
19.     def on_start(self):
    
20.         print('这是SETUP，每次实例化User前都会执行！')
    
21.    
    
22.     @task(1)
    
23.     def getBaidu(self):
    
24.         self.client.get(url="/",verify=False)
    
25.    
    
26.     def on_stop(self):
    
27.         print('这是TEARDOWN，每次销毁User实例时都会执行！')
    
28.    
    
29. 
    
30. if __name__ == "__main__":
    
31.     import os
    
32.     os.system("locust -f testpy.py --headless -u 10 -r 10 -t 2")

复制代码

六、参数化场景一

模拟3个用户并发请求网页，共有100个URL地址，每个虚拟用户都会依次循环加载100个URL地址

1. <font size="3" color="#000000">
   
2. # -*- coding: utf-8 -*-
   
3. """
   
4. @ auth : carl_DJ  来自博主
   
5. @ time : 2020-9-23
   
6. """
   
7. 
   
8. 
   
9. from locust import TaskSet, task, HttpUser
   
10. 
    
11. class UserBehav(TaskSet):
    
12.     def on_start(self):
    
13.         self.index = 0
    
14. 
    
15.     @task
    
16.     def test_visit(self):
    
17.         url = self.user.share_data[self.index]
    
18.         # url = self.locust.share_data[self.index]
    
19.         print('visit url: %s' % url)
    
20.         self.index = (self.index + 1) % len(self.locust.share_data)
    
21.         self.client.get(url)
    
22.         
    
23. class WebsiteUser( HttpUser):
    
24.     host = 'http://www.xxx.com'
    
25.     task_set = task(UserBehav)
    
26.     share_data = ['url1', 'url2', 'url3', 'url4', 'url5']
    
27.     min_wait = 1000
    
28.     max_wait = 15000
    
29. </font>

复制代码

场景二

模拟3用户并发注册账号，共有9个账号，要求注册账号不重复，注册完毕后结束测试

概括：
保证并发测试数据唯一性，不循环取数据
>>>所有并发虚拟用户共享同一份测试数据，并且保证虚拟用户使用的数据不重复；

代码
采用队列

1. <font size="3" color="#000000">
   
2. # -*- coding: utf-8 -*-
   
3. """
   
4. @ auth : carl_DJ 来自博主 这里使用了队列
   
5. @ time : 2020-9-23
   
6. """
   
7. 
   
8. 
   
9. from locust import TaskSet, task, HttpUser
   
10. import queue
    
11. class UserBehav(TaskSet):
    
12.     @task
    
13.     def test_register(self):
    
14.         try:
    
15.             data = self.user.user_data_queue.get()
    
16.             # data = self.locust.user_data_queue.get()
    
17.         except queue.Empty:
    
18.             print('account data run out, test ended.')
    
19.             exit(0)
    
20.         print('register with user: {}, pwd: {}'\
    
21.             .format(data['username'], data['password']))
    
22.         payload = {
    
23.             'username': data['username'],
    
24.             'password': data['password']
    
25.         }
    
26.         self.client.post('/register', data=payload)
    
27.         
    
28. class WebsiteUser(HttpUser):
    
29.     host = 'http://www.xxx.com'
    
30.     task_set = task(UserBehav)
    
31.     user_data_queue = queue.Queue()
    
32.     for index in range(100):
    
33.         data = {
    
34.             "username": "test%04d" % index,
    
35.             "password": "pwd%04d" % index,
    
36.             "email": "test%04d@debugtalk.test" % index,
    
37.             "phone": "186%08d" % index,
    
38.         }
    
39.         user_data_queue.put_nowait(data)
    
40.     min_wait = 1000
    
41.     max_wait = 15000</font>

复制代码

场景三

模拟3个用户并发登录账号，总共有9个账号，要求并发登录账号不相同，但数据可循环使用；

概括：
保证并发测试数据唯一性，循环取数据；
>>>所有并发虚拟用户共享同一份测试数据，保证并发虚拟用户使用的数据不重复，并且数据可循环重复使用

1. <font size="3" color="#000000"># -*- coding: utf-8 -*-
   
2. """
   
3. @ auth : carl_DJ 来自博主 收藏学习
   
4. @ time : 2020-9-23
   
5. """
   
6. 
   
7. 
   
8. from locust import TaskSet, task, HttpUser
   
9. import queue
   
10. class UserBehav(TaskSet):
    
11.     @task
    
12.     def test_register(self):
    
13.         try:
    
14.             data = self.user.user_data_queue.get()
    
15.             # data = self.locust.user_data_queue.get()
    
16.         except queue.Empty:
    
17.             print('account data run out, test ended')
    
18.             exit(0)
    
19.         print('register with user: {0}, pwd: {1}' .format(data['username'], data['password']))
    
20.         payload = {
    
21.             'username': data['username'],
    
22.             'password': data['password']
    
23.         }
    
24.         self.client.post('/register', data=payload)
    
25.         self.user.user_data_queue.put_nowait(data)
    
26.         # self.locust.user_data_queue.put_nowait(data)
    
27. class WebsiteUser(HttpUser):
    
28.     host = 'http://www.xxx.com'
    
29.     task_set = task(UserBehav)
    
30.     user_data_queue = queue.Queue()
    
31.     for index in range(100):
    
32.         data = {
    
33.             "username": "test%04d" % index,
    
34.             "password": "pwd%04d" % index,
    
35.             "email": "test%04d@debugtalk.test" % index,
    
36.             "phone": "186%08d" % index,
    
37.         }
    
38.         user_data_queue.put_nowait(data)
    
39.     min_wait = 1000
    
40.     max_wait = 15000</font>

复制代码

转自：http://t.csdn.cn/IWo2a