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": "",

                   "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进程退出时触发；