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

时间: 2024-11-02 12:10:56

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

Python Locust性能测试框架实践

Locust的介绍 Locust是一个python的性能测试工具,你可以通过写python脚本的方式来对web接口进行负载测试. Locust的安装 首先你要安装python2.6以上版本,而且有pip工具.之后打开命令行,分别安装locustio和pyzmq(命令如下): pip install locustio pip install pyzmq 之后我们就可以写性能测试脚本了. Locust脚本编写 接下来我们拿两个接口做一下测试,编写脚本如下(每一步都有注释).我来解释一下,首先我们要i

python locust 性能测试:locust 关联---提取返回数据并使用

from locust import HttpLocust, TaskSet, taskimport jsonfrom common import readConfig class UserBehavior(TaskSet): # 定义用户行为 def on_start(self): # 当模拟用户开始执行TaskSet类的时候,on_start方法会被调用 pass def get_headers(self): """会员登录""" heade

python locust 性能测试:locust 参数化(list) ---循环取数据,数据可重复使用

from locust import TaskSet, task, HttpLocust class UserBehavior(TaskSet): def on_start(self): # 当模拟用户开始执行TaskSet类的时候,on_start方法会被调用 self.index = 0 @task def test_visit(self): url = self.locust.share_data[ self.index] # 取 self.locust.share_data<等于 Web

python locust 性能测试:HttpSession

官网解释:用于在请求之间执行Web请求和保留(会话)cookie的类(以便能够登录和退出网站):记录每个请求,以便locust可以显示统计信息: from locust import TaskSet, task, HttpLocust, Locust from locust.clients import HttpSession import subprocess import random from common.sec_encrypt import AESEncrypt, md5_time f

python locust 性能测试:locust参数-保证并发测试数据唯一性,循环取数据

from locust import TaskSet, task, HttpLocustimport queue class UserBehavior(TaskSet): @task def test_register(self): try: # get_nowait() 取不到数据直接崩溃:get() 取不到数据会一直等待 data = self.locust.user_data_queue.get_nowait() # 取值顺序 'username': 'test0000'.'usernam

python locust 性能测试:locsut参数化-保证并发测试数据唯一性,不循环取数据

from locust import TaskSet, task, HttpLocustimport queue class UserBehavior(TaskSet): @task def test_register(self): try: # get_nowait() 取不到数据直接崩溃:get() 取不到数据会一直等待 data = self.locust.user_data_queue.get_nowait() # 取值顺序 'username': 'test0000'.'usernam

python locust 性能测试:嵌套

TaskSet类和TaskSequence类可用于嵌套<可以在TaskSequences中嵌套TaskSets,反之亦然>: from locust import TaskSet, task, HttpLocust, TaskSequence, seq_task import subprocess class WebUser(TaskSet): @task(5) def first_task(self): print('执行5次:') @task(2) class IosUser(TaskSe

Python locust性能测试框架模板

locust框架模板 from locust import HttpLocust, TaskSet, task import Queue class UserBehavior(TaskSet): def setup(self): print('task setup') def teardown(self): print('task teardown') def on_start(self): # key = self.login() request_url = "/api/user/getkey

基于python的性能测试工具–locust

现在有很多的性能测试工具,比如说我们熟悉的loadrunner.jmeter.ab.webbench等等,这些工具如果对一个没用过的朋友来说,学习起来比较不容易,但是如果你能看懂python代码,会写就更好了,就可用尝试一下今天的主角Locust,一款基于python的性能测试工具,它的优点是学习起来比较简单,功能完全自定制,使用比较灵活,支持分布式. 所有的性能测试工具都至少包含这3块:               1.压力产生器,也就是可以指定产生多大的压力,多少并发: