当前位置：首页 > news >正文

locust常用类和方法解析

news 2026/4/2 1:43:26

from locust import HttpUser, task, between# 相当于模拟一个用户
class WebUser(HttpUser):# 服务器的地址host = 'http://localhost:8000'# 下一个任务执行之前等待的时间，用于模式用户的思考时间wait_time = between(3, 5)  # 这里使用随机 3，5 秒钟# 定义一个测试任务，类似于一个取样器@task(1)def info(self):# 获取用户信息self.client.get('/info/')@task(10)def addresses(self):# 获取用户地址self.client.get('/addresses/')def on_start(self):# 测试之前执行的操作print('用户登录')auth = {'username': 'root', 'password': 'root1234'}self.client.post('/login/', json=auth)def on_stop(self):# 测试结束执行的操作print('用户退出')self.client.delete('/logout/')

locust导入对象说明

@task: 标记为事务，不定义数值时默认权重为1，权重比越高执行倍率越大
- 通过 @task 装饰一个方法，这个方法表示一个用户可能执行的动作，一个用户会有多种动作;
- 一个用户下一步具体要执行什么动作，是随机的，也就是 locust 会从动作列表中随机选择一个
- @task 接受一个参数，表示权重，数值越大，这个动作被选中的概率就越高
- 所有的task不能是异步方法。
HttpUser：主要用于做http请求，基于reques进行了封装，定义session，locust 中通过 HttpUser 来表示一个用户
- client 的 get,post 等方法传递 url 地址时，不需要携带 http://www.example.com 这个前缀，只需要提供路径就行
FastHttpUser，高并发请求
User，自定义请求
host：配置要测试的目标服务器地址，配置后场景事务访问的接口则不需要再添加该地址
wait_time：下一个任务执行之前等待的时间，用于模拟用户的思考时间。单位：秒
- wait_time = between(1, 5) 间隔是1-5秒随机值
- wait_time = constant(1) 间隔固定1秒
on_start: 每用户初始动作，作用等同于pytest、unittest的setup
on_stop: 每用户结束动作，作用等同于pytest、unittest的teardown
TaskSet或SequentialTaskSet，动作集合，请求有先后顺序时继承SequentialTaskSet类，没有先后顺序，可以使用继承TaskSet类
- 在 locust 中我们通过 TaskSet 类来组合多个动作，然后在 HttpUser 对象中引入 taskset ，引入方式是在 tasks 属性列表中，添加 taskset 类。
- 注意: taskset 相当于一个特殊的动作，如果 locust 进入到 taskset 中执行其中的动作，那么 locust 会一直执行 taskset 中的动作，那么 WebUser 中定义的动作就不会被执行。需要在 taskset 中调用 self.ingterrupt() 来退出，一般可以定义一个动作，在这个动作中调用 self.interrupt()

import uuidfrom locust import HttpUser, task, between, TaskSet# 注册
class RegistTaskSet(TaskSet):# 用户名是否重复@taskdef username_count(self):username = 'zhangsan'self.client.get(f'/usernames/{username}/count/')# 手机号是否重复@taskdef mobile_count(self):mobile = '13712345678'self.client.get(f'/mobiles/{mobile}/count/')# 获取图形验证码@taskdef image_codes(self):image_code_id = str(uuid.uuid4())self.client.get(f'/image_codes/{image_code_id}/')@taskdef stop(self):print('退出当前动作集合')self.interrupt()# 登录
class UserLoginTaskSet(TaskSet):# 嵌套用户注册tasks = [RegistTaskSet]@taskdef username_login(self):auth = {'username': 'root', 'password': 'root1234'}self.client.post('/login/', json=auth)@taskdef stop(self):print('退出当前动作集合')self.interrupt()# 模拟用户
class WebUser(HttpUser):# 包含用户登录tasks = [UserLoginTaskSet]# 服务器的地址host = 'http://localhost:8000'# 下一个任务执行之前等待的时间，用于模式用户的思考时间wait_time = between(3, 5)  # 这里使用随机 3，5 秒钟@taskdef search(self):# 搜索self.client.get('/search/')

断言

需要给请求参数携带 catch_response=True ，这样请求方法就会返回一个上下文管理器

这个上下文管理器返回的是响应对象，我们通过调用响应对象的 failure(消息) 来标记本次请求失败，

通过 catch_response=True 参数来进行断言。

HTTP客户端中使用调用response.failure()或response.success()

with self.client.get("/", catch_response=True) as response:if response.text != "Success":response.failure("Got wrong response")elif response.elapsed.total_seconds() > 0.5:response.failure("Request took too long")

自定义断言

def success_call(name, recvText, total_time):events.request_success.fire(request_type="[Success]",name=name,response_time=total_time,response_length=len(recvText))def fail_call(name, total_time):events.request_failure.fire(request_type="[Fail]",name=name,response_time=total_time,response_length=0,exception=Exception(),)