locust入门系列篇一
目录一、安装
二、常见使用方法
1、直接HttpUser中（常见，一般是代码量少）
2、写在模块中（少见）
3、单独写在taskset中（常见）
三、运行
四、TaskSet与SequentialTaskSet区别
场景一
场景二
五、代码执行顺顺序
六、参数化
场景一
场景二
场景三
一、安装官方：https://docs.locust.io/en/stable/installation.htmlpip3 install locust
# 查看版本
locust -V二、常见使用方法1、直接HttpUser中（常见，一般是代码量少）from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
class Login(HttpUser):
host = 'http://1.72.24.40:23'
# wait_time = constant(3)# 每次请求停顿时间 （思考时间）
wait_time = between(1, 3)
@task
def login(self):
url = '/login'
json = {
"password": "45",
"username": "34",
}
header = {"Content-Type": "application/json;charset=UTF-8"}
self.client.request(method='POST', url=url, json=json, headers=header, name='登录1')2、写在模块中（少见）
from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
@task
def login(param):
url = '/login'
json = {
"password": "45",
"username": "34",
}
header = {"Content-Type": "application/json;charset=UTF-8"}
param.client.request(method='POST', url=url, json=json, headers=header, name='登录1')
@task
def home(param):
url = '/#/login'
param.client.request(method='GET', url=url, name='首页')
class Login(HttpUser):
host = 'http://1.72.24.40:23'
# wait_time = constant(3)# 每次请求停顿时间 （思考时间）
wait_time = between(1, 3)
tasks = 3、单独写在taskset中（常见）
from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
class TaskTest(TaskSet):
@task
def login(self):
url = '/login'
json = {
"password": "rewr",
"username": "rewr",
}
header = {"Content-Type": "application/json;charset=UTF-8"}
self.client.request(method='POST', url=url, json=json, headers=header, name='登录')
class Login(HttpUser):
host = 'http://s2.x.x.x:x'
# wait_time = constant(3)# 每次请求停顿时间 （思考时间）
wait_time = between(1, 3)
tasks =
三、运行
locust -f 文件路径+文件名.py 浏览器输入：http://localhost:8089/
四、TaskSet与SequentialTaskSet区别 继承TaskSet类，那么对应里面的方法是并行执行的，继承SequentialTaskSet类，那么对应里面的方法是串行执行的。场景一
例如：登录一次，添加50次，那么可以考虑使用继承SequentialTaskSet类，然后在装饰器task当中进行设置
场景二
配合on_start、on_stop（这里例子没有使用到）使用askSet然后通过配置task。on_start、on_stop是每个用户开始和结束的时候都会执行一次。
模拟100个用户，先登录系统，然后90%执行A操作，10%执行B操作# -*- coding: utf-8 -*-
# @Time : 2021/6/14 8:17
# @Project : virlocustdemo
# @Author: testing
# @File : locus2ss.py
# @Software: PyCharm
from locust import HttpUser, constant, between, task, TaskSet, SequentialTaskSet
class TaskTest(TaskSet):
"""
这里继承的是串行执行的类
"""
token = None
# 用户执行task前调用
def on_start(self):
url = '/XX/XX/login'
json = {
"password": "888X23338",
"username": "XXXX"
}
header = {"Content-Type": "application/json}
try:
with self.client.post(url=url, json=json, headers=header, name='登录') as response:
self.token = 'Bearer' + ' ' + res['data']['access_token']
except Exception as e:
print("错误信息为", e)
# 用户执行task后调用
def on_stop(self):
print("用户结束")
@task(90)
def query_users(self):
url = '/DSAD/userMDSADe/DS'
json = {
"pageIndex": "1",
"pageSize": "200"
}
header = {
"Content-Type": "application/json,
"Authorization": self.token}
try:
self.client.get(url=url, params=json, headers=header, name='A')
except Exception as e:
print("错误信息为", e)
@task(10)
def query_oErg_tEWee(self):
url = '/uWEW/usEWEWe/orEWEree'
json = {"enRdLeERvTel": "1"}
header = {
"Content-Type": "application/json,
"Authorization": self.token}
try:
self.client.get(url=url, params=json, headers=header, name='B')
except Exception as e:
print("错误信息为", e)
class Login(HttpUser):
host = 'http://X2.122.123.20:138'
# wait_time = constant(3)# 每次请求停顿时间 （思考时间）
wait_time = between(1, 3)
tasks = 五、代码执行顺顺序测试项目执行前会执行一次SETUP，测试项目结束后会执行一次TEARDOWN。import time
from locust import HttpUser,task,between,events
import urllib3
urllib3.disable_warnings()
@events.test_start.add_listener
def on_test_start(**kwargs):
print('===测试最开始提示===')
@events.test_stop.add_listener
def on_test_stop(**kwargs):
print('===测试结束了提示===')
class TestTask(HttpUser):
wait_time = between(1, 5)
host = 'https://www.baidu.com'
def on_start(self):
print('这是SETUP，每次实例化User前都会执行！')
@task(1)
def getBaidu(self):
self.client.get(url="/",verify=False)
def on_stop(self):
print('这是TEARDOWN，每次销毁User实例时都会执行！')
if __name__ == "__main__":
import os
os.system("locust -f testpy.py --headless -u 10 -r 10 -t 2")六、参数化场景一 模拟3个用户并发请求网页，共有100个URL地址，每个虚拟用户都会依次循环加载100个URL地址
<font size="3" color="#000000">
# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ来自博主
@ time : 2020-9-23
"""
from locust import TaskSet, task, HttpUser
class UserBehav(TaskSet):
def on_start(self):
self.index = 0
@task
def test_visit(self):
url = self.user.share_data
# url = self.locust.share_data
print('visit url: %s' % url)
self.index = (self.index + 1) % len(self.locust.share_data)
self.client.get(url)
class WebsiteUser( HttpUser):
host = 'http://www.xxx.com'
task_set = task(UserBehav)
share_data = ['url1', 'url2', 'url3', 'url4', 'url5']
min_wait = 1000
max_wait = 15000
</font>
场景二模拟3用户并发注册账号，共有9个账号，要求注册账号不重复，注册完毕后结束测试概括：
保证并发测试数据唯一性，不循环取数据
>>>所有并发虚拟用户共享同一份测试数据，并且保证虚拟用户使用的数据不重复；代码
采用队列 <font size="3" color="#000000">
# -*- coding: utf-8 -*-
"""
@ auth : carl_DJ 来自博主 这里使用了队列
@ time : 2020-9-23
"""
from locust import TaskSet, task, HttpUser
import queue
class UserBehav(TaskSet):
@task
def test_register(self):
try:
data = self.user.user_data_queue.get()
# data = self.locust.user_data_queue.get()
except queue.Empty:
print('account data run out, test ended.')
exit(0)
print('register with user: {}, pwd: {}'\
.format(data['username'], data['password']))
payload = {
'username': data['username'],
'password': data['password']
}
self.client.post('/register', data=payload)
class WebsiteUser(HttpUser):
host = 'http://www.xxx.com'
task_set = task(UserBehav)
user_data_queue = queue.Queue()
for index in range(100):
data = {
"username": "test%04d" % index,
"password": "pwd%04d" % index,
"email": "test%04d@debugtalk.test" % index,
"phone": "186%08d" % index,
}
user_data_queue.put_nowait(data)
min_wait = 1000
max_wait = 15000</font>场景三模拟3个用户并发登录账号，总共有9个账号，要求并发登录账号不相同，但数据可循环使用；概括：
保证并发测试数据唯一性，循环取数据；
>>>所有并发虚拟用户共享同一份测试数据，保证并发虚拟用户使用的数据不重复，并且数据可循环重复使用<font size="3" color="#000000"># -*- coding: utf-8 -*-
"""
@ auth : carl_DJ 来自博主 收藏学习
@ time : 2020-9-23
"""
from locust import TaskSet, task, HttpUser
import queue
class UserBehav(TaskSet):
@task
def test_register(self):
try:
data = self.user.user_data_queue.get()
# data = self.locust.user_data_queue.get()
except queue.Empty:
print('account data run out, test ended')
exit(0)
print('register with user: {0}, pwd: {1}' .format(data['username'], data['password']))
payload = {
'username': data['username'],
'password': data['password']
}
self.client.post('/register', data=payload)
self.user.user_data_queue.put_nowait(data)
# self.locust.user_data_queue.put_nowait(data)
class WebsiteUser(HttpUser):
host = 'http://www.xxx.com'
task_set = task(UserBehav)
user_data_queue = queue.Queue()
for index in range(100):
data = {
"username": "test%04d" % index,
"password": "pwd%04d" % index,
"email": "test%04d@debugtalk.test" % index,
"phone": "186%08d" % index,
}
user_data_queue.put_nowait(data)
min_wait = 1000
max_wait = 15000</font>转自：http://t.csdn.cn/IWo2a
页:
[1]
