python locust 性能測試：HOOKS<鈎子方法>

本文轉載自查看原文 2019-01-01 16:02 651 locust/ python

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