python locust 性能测试:HOOKS<钩子方法>

为locust中不同类型的事件,提供的钩子方法;

from locust import TaskSet, task, events, Locust
from locust.clients import HttpSession
import subprocess
import random, traceback
from common.sec_encrypt import AESEncrypt, md5_time
from common import read_config
from common.logger import print_file

def on_request_success(request_type, name, response_time, response_length):
    print_file(
        ‘success  Type: {}, Name: {}, Time: {}ms, length: {}‘.format(request_type, name, response_time,
                                                                     response_length))

def on_request_failure(request_type, name, response_time, exception):
    print_file(
        ‘failure  Type: {}, Name: {}, Time: {}ms, exception: {}‘.format(request_type, name, response_time,
                                                                        exception))

def on_locust_error(locust_instance, exception, tb):
    print_file(
        "error  locust_instance: {}, exception: {}, traceback: {}".format(locust_instance, exception,
                                                                          traceback.format_tb(tb)))

def on_hatch_complete(user_count):
    print_file("User  Locust have generate {} users".format(user_count))

def on_quitting():
    print_file("Locust is quiting!")

events.request_success += on_request_success
events.request_failure += on_request_failure
events.locust_error += on_locust_error
events.hatch_complete += on_hatch_complete
events.quitting += on_quitting

class WebUser(TaskSet):
    def on_start(self):
        self.session = HttpSession(‘http://localhost:8000/api‘)
        self.r = random.randint(10000, 99999)

    @task(2)
    def user_auth(self):
        """用户认证"""
        auth_user = (‘lixiaofeng‘, ‘fengzi802300‘)
        res = self.session.get(‘/sec_get_event_list/‘, params={‘eid‘: self.r}, auth=auth_user,
                               name=‘/sec_get_event_list/?eid=[eid]‘)
        print_file(res.json())
        # assert res.json()[‘status‘] == 200, ‘用户认证接口请求失败!‘

    @task(1)
    def aes_api(self):
        payload = {‘name‘: ‘187187011{}‘.format(self.r)}
        # 加密
        encodeed = AESEncrypt(payload).encrypt_data()
        # print(encodeed)
        res = self.session.get(‘/sec_get_guest_list/‘, params={‘data‘: encodeed},
                               name=‘/sec_get_guest_list/?data=[aes]‘)
        print_file(res.text)
        # assert res.json()[‘status‘] == 200, ‘AES接口加密请求失败!‘

    @task(2)
    def md5_api(self):
        payload = {"eid": "{}".format(self.r), "name": "一加8手机发布会{}".format(str(self.r)), "limit": 2000, "status": "1",
                   "address": "深圳宝体会展中心",
                   "start_time": "2019-09-15 22:40:00", ‘time‘: ‘‘, ‘sign‘: ‘‘}
        payload = md5_time(payload)
        res = self.session.post(‘/sec_add_event/‘, data=payload, name=‘/sec_add_event/?data=[md5]‘)
        print_file(res.text)
        # assert res.json()[‘status‘] == 200, ‘md5接口加密请求失败!‘

    @task(2)
    def add_guest(self):
        """添加嘉宾"""
        payload = {‘eid‘: ‘{}‘.format(self.r), ‘realname‘: ‘赵小刀{}‘.format(self.r),
                   ‘phone‘: ‘187011{}‘.format(self.r),
                   ‘email‘: ‘187011{}@163.com‘.format(self.r)}
        res = self.session.post(‘/add_guest/‘, data=payload, name=‘/add_guest/?data=[payload]‘)
        print_file(res.text)

class Run(Locust):
    # host = ‘http://localhost:8000/api‘
    task_set = WebUser
    max_wait = 6000
    min_wait = 3000

# ‘locust -f D:\Interface_framework_Beauty\locust_files\locust_api.py --no-web -c 100 -r 10 --run-time {}s‘.format(read_config.time)
def run():
    subprocess.check_call(
        ‘locust -f G:\Interface_framework_pytest\locust_files\\locust_test.py --no-web -c 1000 -r 100 --run-time {}s‘.format(
            read_config.time))

if __name__ == ‘__main__‘:
    run()

1.request_success 成功完成请求后触发;

  参数:

    request_tye:请求类型;

    name:被调用url路径;

    response_time:响应时间,ms;

    response_length:响应内容长度;

2.request_failure 请求失败时触发;

  参数:

    request_tye:请求类型;

    name:被调用url路径;

    response_time:响应时间,ms;

    exception:抛出的异常;

3.locust_eror 在执行locust类内发生异常时触发;

  参数:

    locust_instance:发生异常的locust实例;

    exception:抛出的异常;

    tb:回溯的对象;

4.hatch_complete 生成所有locust用户时触发;

  参数:

    user_count:生成的用户数;

5.quitting locust进程退出时触发;



原文地址:https://www.cnblogs.com/changqing8023/p/10205038.html

时间: 01-01

python locust 性能测试:HOOKS<钩子方法>的相关文章

python常用数据类型内置方法介绍

熟练掌握python常用数据类型内置方法是每个初学者必须具备的内功. 一.整型 a = 100 a.xxx() class int(object): def bit_length(self): ##如果将某个整数用2进制表示,返回这个2进制所占bit位数. return 0 def conjugate(self, *args, **kwargs): ##共轭复数 @classmethod # known case def from_bytes(cls, bytes, byteorder, *ar

python 的__del__删除器方法

遇到了python关于del的小问题,又引发了一些问题,就简单的记录下,如果有不同意见,欢迎一起讨论~ python类似java是有垃圾回收机制的语言,所以我们不需要像c++一样通过析构函数来手动释放内存, 但是python也同样提供了__del__释放方法.当一个对象的引用计数为0时,会被自动调用,那么先来说说 引用计数吧. 1.引用计数 #!/usr/bin/env python #coding:UTF-8 import sys class MyClass(object): def __in

Hooks——钩子概览

(资料源于MSDN,本文仅对其进行翻译.批注.其链接为:http://msdn.microsoft.com/en-us/library/windows/desktop/ms644959%28v=vs.85%29.aspx本文链接:http://blog.csdn.net/wlsgzl/article/details/37648721转载请注明出处并保持文章的完整性.) Hooks Overview A hook is a mechanism by which an application can

python直接生成exe的方法

一.简介 py2exe是一个将python脚本转换成windows上的可独立执行的可执行程序(*.exe)的工具,这样,你就可以不用装python而在windows系统上运行这个可执行程序.py2exe已经被用于创建wxPython,Tkinter,Pmw,PyGTK,pygame,win32com client和server,和其它的独立程序.py2exe是发布在开源许可证下的.目前只有python2.x的才有对应版本的.二.安装py2exe 从http://prdownloads.sourc

【转】python中List的sort方法(或者sorted内建函数)的用法

原始出处:http://gaopenghigh.iteye.com/blog/1483864 python列表排序 简单记一下python中List的sort方法(或者sorted内建函数)的用法. 关键字: python列表排序 python字典排序 sorted List的元素可以是各种东西,字符串,字典,自己定义的类等. sorted函数用法如下: Python代码   sorted(data, cmp=None, key=None, reverse=False) 其中,data是待排序数

Python文件遍历二种方法

分享下有关Python文件遍历的两种方法,使用的OS模块的os.walk和os.listdir实现. 关于Python的文件遍历,大概有两种方法,一种是较为便利的os.walk(),还有一种是利用os.listdir()递归遍历.方法一:利用os.walkos.walk可以自顶向下或者自底向上遍历整个文件树,然后返回一个含有3个元素的tuple,(dirpath, dirnames, filenames).注意,os.walk()会返回一个generater,所以调用的时候一定要放到for循环中

python求crc32值的方法

python求crc32值的方法 在python代码中求CRC值,要import binasciibinascii.crc32(v) 求出了v的crc32值,这是一个long型,形如-1456387L,把这个值&0xffffffff得到的值形如48a213L的形式.然后,把这个值用16进制表示出来.www.jbxue.com例子: def _crc32(self, v): """ Generates the crc32 hash of the v. @return:

Python读写excel表格的方法二

目的:实现用python的另一种方法做excel的读取.新增操作. 环境:ubuntu 16.04  Python 3.5.2 情景:之前介绍了一种操作excel文件的方法(私链),现在使用另一种方法读写excel文件,一次性读出或写入,读写也很方便,读出为有序字典,像二维数组,需要用到3个第三方模块,pyexcel-io, pyexcel-xls, xlrd,主要导入pyexcel-xls和collections,pyexcel-xls基于xlrd和pyexcel-io,所以支持xls和xls

[Python] 正确复制列表的方法

new = old[:] Python老鸟都知道以上代码是什么意思.它复制列表old到new.它对于新手来说是种困惑而且应该避免使用这种方法.不幸的是[:]标记法被广泛使用,可能是Python程序员不知道更好的列表复制法吧. 首先我们需要了解Python是如何管理对象和变量.Python没有C语言中的变量.在C语言中,变量不止是个名字,它是字节集合并真实存在于内存某个位置上.而在Python中,变量仅仅是指向对象的标签. 看看以下语句: a = [1, 2, 3] 它表示我们创建了一个指引指向列

Python常用异常及处理方法

Python 异常处理 python提供了两个非常重要的功能来处理python程序在运行中出现的异常和错误.你可以使用该功能来调试python程序. 异常处理: 本站Python教程会具体介绍. 断言(Assertions):本站Python教程会具体介绍. python标准异常 异常名称 描述 BaseException 所有异常的基类 SystemExit 解释器请求退出 KeyboardInterrupt 用户中断执行(通常是输入^C) Exception 常规错误的基类 StopIter