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

Python Locust性能测试框架实践

[本文出自天外归云的博客园] Locust的介绍 Locust是一个python的性能测试工具,你可以通过写python脚本的方式来对web接口进行负载测试. Locust的安装 首先你要安装pyth ...

python locust 性能测试：locust安装和一些参数介绍

安装参考 https://www.cnblogs.com/fnng/p/6081798.html ps:python3.7暂不支持locust:python3安装建 ...

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

from locust import HttpLocust, TaskSet, taskimport jsonfrom common import readConfig class UserBehav ...

python locust 性能测试：HttpSession

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

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

from locust import TaskSet, task, HttpLocust class UserBehavior(TaskSet): def on_start(self): # 当模拟用 ...

python locust 性能测试：嵌套

TaskSet类和TaskSequence类可用于嵌套: from locust import TaskSet, tas ...

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

from locust import TaskSet, task, HttpLocustimport queue class UserBehavior(TaskSet): @task def test ...

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

from locust import TaskSet, task, HttpLocustimport queue class UserBehavior(TaskSet): @task def test ...

Python locust性能测试框架模板

locust框架模板 from locust import HttpLocust, TaskSet, task import Queue class UserBehavior(TaskSet): de ...

随机推荐

oracle唯一索引与普通索引的区别和联系以及using index用法

oracle唯一索引与普通索引的区别和联系 区别:唯一索引unique index和一般索引normal index最大的差异是在索引列上增加一层唯一约束.添加唯一索引的数据列可以为空,但是只要尊在数 ...

系统不支持curl

解决安装pinphp时出现的“系统不支持curl!”   今天在本机安装phppin开源程序时,提示“系统不支持curl!”错误. 由于我本机是UBUNTU系统,所以直接通过apt-get进行安装. ...

ES5严格模式(Strict mode)

严格模式(Strict mode)是由ECMA-262规范定义的新兴JavaScript标准,第五版发布于2009年12月.旨在改善错误检查功能并且标识可能不会延续到未来JavaScript版本的脚本 ...

在c++程序中执行DOS命令

转自博客:http://blog.csdn.net/ypist/article/details/8485049 #1,system()方式 在C盘根目录下新建文件夹,名称为12: system(&qu ...

thinkPhp 3.1.3的验证码无法显示的问题

Image帮助类的output方法中,在下面的代码 header("Content-type: image/" . $type); 前增加代码: ob_end_clean();

JavaScript 运行机制详解：再谈Event Loop