51Testing软件测试论坛

标题: locust入门系列篇一 [打印本页]
作者: 梦幻小丑灯    时间: 2022-8-11 15:12
标题: locust入门系列篇一
目录
一、安装

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

三、运行

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

五、代码执行顺顺序

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

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

  1. pip3 install locust

  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')

  12. @task
  13. def home(param):
  14.     url = '/#/login'
  15.     param.client.request(method='GET', url=url, name='首页')

  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


  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='登录')




  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是每个用户开始和结束的时候都会执行一次。

  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


  10. class TaskTest(TaskSet):
  11.     """
  12.     这里继承的是串行执行的类
  13.     """
  14.     token = None

  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)


  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()

  6. @events.test_start.add_listener
  7. def on_test_start(**kwargs):
  8.     print('===测试最开始提示===')

  10. @events.test_stop.add_listener
  11. def on_test_stop(**kwargs):
  12.     print('===测试结束了提示===')


  15. class TestTask(HttpUser):

  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.    

  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. """


  9. from locust import TaskSet, task, HttpUser

  11. class UserBehav(TaskSet):
  12.     def on_start(self):
  13.         self.index = 0

  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. """


  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. """


  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




欢迎光临 51Testing软件测试论坛 (http://bbs.51testing.com/) Powered by Discuz! X3.2