深聊性能测试，从入门到放弃之：Locust性能自动化(六)自定义生成负载图形形状

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-10-22
"""
from locust import LoadTestShape
class CustomShape(LoadTestShape):
  #设置时限
  time_limit = 600
  #设置产生率
  spawn_rate = 20
  def tick(self):
  '''
  设置 tick()函数
  并在tick()里面调用 get_run_time()方法
  '''
  #调用get_run_time()方法
  run_time = self.get_run_time()
  #做比较,运行时间在 10分钟之内，则继续执行
  if run_time < self.time_limit:
    #将用户数四舍五入到最接近的百分值
    uesr_count =round(run_time,-2)
    #返回user_count,spawn_rate这两个参数
    return (user_count,spawn_rate)
  return None

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-10-22
"""
import math
from locust import LoadTestShape
from locust import HttpUser,TaskSet,task,constant
class UserTasks(TaskSet):
  #设置task
    @task
    def get_root(self):
        self.client.get('/')
class WebsiteUser(HttpUser):
    wait_time = constant(0.5)
    tasks = [UserTasks]
class DoubleWave(LoadTestShape):
    '''
    自定义一个双波形图形，
    模拟在某两个时间点的最高值
    参数解析:
        min_users ： 最小用户数
        peak_one_users ： 用户在第一个峰值
        peak_two_users ： 用户在第二个峰值
        time_limit ： 测试执行总时间
    '''
    # 最小用户数
    min_users = 20
    #第一个峰值的用户数
    peak_one_users = 60
    #第二个峰值的用户数
    peak_two_users = 40
    #测试执行时间
    time_limit = 600
    def tick(self):
        #将get_run_time 四舍五入
        run_time = round(self.get_run_time)
        if run_time < self.time_limit:
            user_count = (
            (self.peak_one_users - self.min_users)
            # *math.e **  - (((run_time / （self.time_limit / 10 *2 / 3)) - 5) ** 2）
             * math.e ** -(((run_time / (self.time_limit / 10 * 2 / 3)) - 5) ** 2)
            + (self.peak_two_users - self.min_users)
            * math.e ** -(((run_time / (self.time_limit / 10 * 2 / 3)) - 10) ** 2)
            + self.min_users
            )
            return (round(user_count),round(user_count))
        else:
            return None

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-10-22
"""
from locust import LoadTestShape
from locust import HttpUser,TaskSet,task,constant
class UserTasks(TaskSet):
    @task
    def get_root(self):
        self.client.get('/')
class WebsiteUser(HttpUser):
    wait_time = constant(0.5)
    tasks = [UserTasks]
class StagesShape(LoadTestShape):
    '''
    在不同的阶段 具有不同的用户数和 产生率的 图形形状
    参数解析:
        stages ：字典列表，每个字典都具有下列这些键值的阶段:
            duration -- 持续时间，  经过多少秒后，进入到下个阶段
            users -- 总用户数
            spawn_rate -- 产生率，即每秒启动/停止的用户数
            stop -- 可以在特定阶段停止测试的值
        stop_at_end -- 可以在所有阶段设置运行后停止
    '''
    stages = [
        {"duration": 60, "users": 10, "spawn_rate": 10},
        {"duration": 100, "users": 50, "spawn_rate": 10},
        {"duration": 180, "users": 100, "spawn_rate": 10},
        {"duration": 220, "users": 30, "spawn_rate": 10},
        {"duration": 230, "users": 10, "spawn_rate": 10},
        {"duration": 240, "users": 1, "spawn_rate": 1}
    ]
    def tick(self):
        run_time = self.get_run_time()
        for stage in self.stages:
            if run_time < stages['duration']:
                tick_data = (stage['users'],stage['spawn_rate'])
                return tick_data
        return None

# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ
@ time : 2020-10-22
"""
from locust import LoadTestShape
from locust import HttpUser,TaskSet,task,constant
class UserTasks(TaskSet):
    @task
    def get_root(self):
        self.client.get('/')
class WebsiteUser(HttpUser):
    wait_time = constant(0.5)
    tasks = [UserTasks]
class StepLoadShaper(LoadTestShape):
    '''
    逐步加载实例
    参数解析：
        step_time -- 逐步加载时间
        step_load -- 用户每一步增加的量
        spawn_rate -- 用户在每一步的停止/启动 
        time_limit -- 时间限制
    '''
    setp_time = 30
    setp_load = 10
    spawn_rate = 10
    time_limit =  600
    def tick(self):
        run_time = self.get_run_time()
        if run_time > self.time_limit:
            return None
        current_step = math.floor(run_time /self.setp_time) +1
        return(current_step * self.setp_load,self.spawn_rate)