当前位置:网站首页>Locust performance test 4 (custom load Policy)
Locust performance test 4 (custom load Policy)
2022-07-07 09:08:00 【Song_ Lun】
Preface
Sometimes we need a fully customized load test , This cannot be achieved by simply setting or changing the number of users and the brush out rate . for example , You may want to generate a load spike or rise or fall at a custom time . By using LoadTestShape
class , You can completely control the user count and generation rate at any time .
Time based peak strategy
There is such a need , A login interface has 10 Users log in at the same time , And continue 180 second , So we need to set it like this ? I'm going to use LoadTestShape
class
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/11/26 10:11 In the morning
# @Name : peilun
# @File : Testshape_test.py
from locust import LoadTestShape, task, HttpUser, constant
import json
""" Custom load policy Generate every second 5 Users , The duration of the 180s """
class MyUser(HttpUser):
# Request interval
wait_time = constant(1)
host = 'https://www.baidu.com'
url = '/api/user/login'
headers = {
'Content-Type': 'application/json'}
@task
def on_start(self):
global Authorization
# Login is performed only once obtain token
print(" User initialization -- Log in and get token")
self.data = {
"username":"123456","password":"123456"}
respon = self.client.post(self.url, headers = self.headers, data=json.dumps(self.data), name=' The user login ', verify=False, timeout=10)
resp_dict = respon.json()
# print(f' The response data is :{resp_dict}')
if respon.status_code == 200:
# print(resp_dict['msg'])
print(f' The response data is :{
resp_dict}')
Authorization = respon.headers['Authorization']
return Authorization
else:
respon.failure(resp_dict['msg'])
# Generate every second 10 Users , The duration of the 180s
class MyCustomShape(LoadTestShape):
# Set the duration of pressure measurement , Unit second
time_limit = 180
# Start every second / Number of users stopped
spawn_rate = 10
def tick(self):
""" Returns a tuple , Contains two values : user_count -- Total users spawn_rate -- Start every second / Number of users stopped return None when , Stop the load test """
# Get the execution time of pressure test
run_time = self.get_run_time()
# The operation duration is within the maximum duration of pressure measurement , Then continue
if run_time < self.time_limit:
user_count = round(run_time, -1)
return user_count, self.spawn_rate
return None
if __name__ == '__main__':
import os
os.system("locust -f Testshape_test.py")
Step based load policy
Every time 30 Second increase 5 Users , continued 10 minute
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/11/26 10:11 In the morning
# @Name : peilun
# @File : Testshape_test2.py
from locust import LoadTestShape, task, HttpUser, constant
import json
import math
""" Custom load policy Every time 30 Second increase 5 Users , continued 10 minute """
class MyUser(HttpUser):
# Request interval
wait_time = constant(1)
host = 'https://www.baidu.com'
url = '/api/user/login'
headers = {
'Content-Type': 'application/json'}
@task
def on_start(self):
global Authorization
# Login is performed only once obtain token
print(" User initialization -- Log in and get token")
self.data = {
"username":"123456","password":"123456"}
respon = self.client.post(self.url, headers = self.headers, data=json.dumps(self.data), name=' The user login ', verify=False, timeout=10)
resp_dict = respon.json()
# print(f' The response data is :{resp_dict}')
if respon.status_code == 200:
# print(resp_dict['msg'])
print(f' The response data is :{
resp_dict}')
Authorization = respon.headers['Authorization']
return Authorization
else:
respon.failure(resp_dict['msg'])
# Generate every second 10 Users , The duration of the 180s
class MyCustomShape(LoadTestShape):
""" step_time -- Time between steps step_load -- Users increase the number at each step spawn_rate -- The user stops every second in every step / Number of starts time_limit -- The time limit , In seconds """
step_time = 30
step_load = 5
spawn_rate = 10
time_limit = 600
def tick(self):
# Get the time to perform the pressure test
run_time = self.get_run_time()
# The running time is beyond the time limit , Do not perform
if run_time > self.time_limit:
return None
current_step = math.floor(run_time / self.step_time) + 1
return current_step * self.step_load, self.spawn_rate
if __name__ == '__main__':
import os
os.system("locust -f Testshape_test2.py")
Load policy based on time period
front 10s and 10-20s The number of users is 10;20-50s The number of users is 50;50-80s The number of users is 100;80s The number of subsequent users is 30
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time : 2021/11/26 10:11 In the morning
# @Name : peilun
# @File : Testshape_test3.py
from locust import LoadTestShape, task, HttpUser, constant
import json
""" Custom load policy front 10s and 10-20s The number of users is 10;20-50s The number of users is 50;50-80s The number of users is 100;80s The number of subsequent users is 30 """
class MyUser(HttpUser):
# Request interval
wait_time = constant(1)
host = 'https://www.baidu.com'
url = '/api/user/login'
headers = {
'Content-Type': 'application/json'}
@task
def on_start(self):
global Authorization
# Login is performed only once obtain token
print(" User initialization -- Log in and get token")
self.data = {
"username":"123456","password":"123456"}
respon = self.client.post(self.url, headers = self.headers, data=json.dumps(self.data), name=' The user login ', verify=False, timeout=10)
resp_dict = respon.json()
# print(f' The response data is :{resp_dict}')
if respon.status_code == 200:
# print(resp_dict['msg'])
print(f' The response data is :{
resp_dict}')
Authorization = respon.headers['Authorization']
return Authorization
else:
respon.failure(resp_dict['msg'])
class MyCustomShape(LoadTestShape):
""" duration -- How many seconds to enter the next stage users -- The number of users spawn_rate -- Start every second / Number of users stopped """
stages = [
{
"duration": 10, "users": 10, "spawn_rate": 10},
{
"duration": 20, "users": 50, "spawn_rate": 10},
{
"duration": 50, "users": 100, "spawn_rate": 10},
{
"duration": 80, "users": 30, "spawn_rate": 10}
]
def tick(self):
run_time = self.get_run_time()
for stage in self.stages:
if run_time < stage["duration"]:
tick_data = (stage["users"], stage["spawn_rate"])
return tick_data
if __name__ == '__main__':
import os
os.system("locust -f Testshape_test3.py")
边栏推荐
- Ppt template and material download website (pure dry goods, recommended Collection)
- Newly found yii2 excel processing plug-in
- Hard core sharing: a common toolkit for hardware engineers
- 2021 year end summary
- How to use Arthas to view class variable values
- Output a spiral matrix C language
- selenium自动化集成,八年测试经验软测工程师,一篇文章带你学懂
- LED模拟与数字调光
- Markdown编辑器Editor.md插件的使用
- Unityshader introduction essentials personal summary -- Basic chapter (I)
猜你喜欢
Simple use of Xray
Full link voltage test of the e-commerce campaign Guide
Recommended by Alibaba P8, the test coverage tool - Jacobo is very practical
How long does the PMP usually need to prepare for the exam in advance?
Nanjing commercial housing sales enabled electronic contracts, and Junzi sign assisted in the online signing and filing of housing transactions
Esp32-ulp coprocessor low power mode RTC GPIO interrupt wake up
How to use Arthas to view class variable values
Ppt template and material download website (pure dry goods, recommended Collection)
PMP Exam details after the release of the new exam outline
H3C VXLAN配置
随机推荐
Count the number of words C language
JVM 垃圾回收 详细学习笔记(二)
Vagrant failed to mount directory mount: unknown filesystem type 'vboxsf'
PMP Exam Preparation experience, seek common ground while reserving differences, and successfully pass the exam
Recommended by Alibaba P8, the test coverage tool - Jacobo is very practical
Personal deduction topic classification record
Reading notes of pyramid principle
外部中断实现按键实验
PMP certificate preparation experience sharing
Enterprise manager cannot connect to the database instance
【ChaosBlade:节点 CPU 负载、节点网络延迟、节点网络丢包、节点域名访问异常】
Serial port experiment - simple data sending and receiving
What are the conditions for applying for NPDP?
Hard core sharing: a common toolkit for hardware engineers
Output a spiral matrix C language
模拟卷Leetcode【普通】1567. 乘积为正数的最长子数组长度
模拟卷Leetcode【普通】1557. 可以到达所有点的最少点数目
5A summary: seven stages of PMP learning
面试题:高速PCB一般布局、布线原则
LeetCode 715. Range 模块