Класс HttpUser, декоратор @task, веса задач, wait_time, self.client, catch_response и валидация.
HttpUser— основной строительный блок Locust. Понимание его API позволяет описать любой реалистичный сценарий: от простого GET до авторизованных POST-запросов с проверкой результата.
Каждый экземпляр HttpUser получает self.client — объект класса HttpSession, который наследует от requests.Session. Это значит, что все привычные методы requests (get, post, put, delete, patch, head) доступны напрямую:
from locust import HttpUser, task, between
class ApiUser(HttpUser):
wait_time = between(0.5, 2)
@task
def get_user(self):
self.client.get("/api/users/1")
@task
def create_user(self):
self.client.post(
"/api/users",
json={"name": "Alice", "email": "alice@example.com"},
headers={"Authorization": "Bearer token123"}
)Поскольку HttpSession наследует от requests.Session, cookies и заголовки сохраняются между запросами в рамках одного пользователя. Это удобно для авторизации: один раз залогиниться в on_start, а дальше все запросы пойдут с cookie-сессией.
on_start вызывается один раз при создании виртуального пользователя, on_stop — при его остановке. Здесь удобно делать логин:
class AuthenticatedUser(HttpUser):
wait_time = between(1, 2)
def on_start(self):
response = self.client.post("/auth/login", json={
"username": "testuser",
"password": "secret"
})
# requests.Session автоматически сохранит Set-Cookie из ответа
@task
def view_dashboard(self):
self.client.get("/dashboard")
def on_stop(self):
self.client.post("/auth/logout")Важно: on_start выполняется в greenlet каждого пользователя, поэтому несколько пользователей логинятся параллельно. Это именно то поведение, которое нужно имитировать.
Locust предоставляет три готовые функции для wait_time:
Функция between(min, max) генерирует случайное число секунд в диапазоне. Используется чаще всего, так как имитирует нерегулярное поведение реальных пользователей:
from locust import between
wait_time = between(1, 5)Функция constant(seconds) всегда ждёт ровно указанное число секунд. Полезна, когда нужна точная и предсказуемая нагрузка:
from locust import constant
wait_time = constant(1)Функция constant_throughput(target_rps) адаптирует паузу так, чтобы каждый пользователь выполнял target_rps задач в секунду. Если задача выполнялась быстро — пауза длиннее, если медленно — короче:
from locust import constant_throughput
wait_time = constant_throughput(2) # 2 задачи в секунду на пользователяПо умолчанию Locust группирует статистику по URL. Динамические URL с ID засоряют таблицу статистики:
/api/users/1
/api/users/2
/api/users/3
Чтобы объединить их под одним именем, передайте параметр name:
@task
def get_user(self):
user_id = random.randint(1, 1000)
self.client.get(f"/api/users/{user_id}", name="/api/users/:id")Теперь все запросы к разным пользователям будут отображаться как одна строка /api/users/:id в статистике.
По умолчанию Locust считает запрос успешным, если HTTP-статус не 4xx/5xx. Но бизнес-логика может требовать более тонкой проверки: например, статус 200, но тело ответа содержит {"error": "not found"}.
Контекстный менеджер catch_response=True позволяет вручную управлять статусом:
@task
def check_health(self):
with self.client.get("/health", catch_response=True) as response:
if response.status_code == 200:
data = response.json()
if data.get("status") != "ok":
response.failure(f"Unhealthy: {data.get('status')}")
else:
response.success()
else:
response.failure(f"HTTP {response.status_code}")Внутри контекста доступны два метода:
response.success() — пометить запрос как успешныйresponse.failure(message) — пометить как упавший с произвольным сообщениемЕсли вы не хотите, чтобы запрос попал в статистику вообще (например, preflight OPTIONS), используйте response.failure() с флагом catch_response=True и не вызывайте ни success(), ни failure() — в таком случае Locust использует HTTP-статус по умолчанию.
Иногда нужно измерить время не одного запроса, а целого пользовательского сценария (например, «добавить товар в корзину и оформить заказ»). Для этого используется менеджер событий вручную или RequestEventMixin. Самый простой способ — измерить время вручную и записать через self.environment.events.request.fire().
Более практичный паттерн — использовать параметр name для логической группировки:
@task
def checkout_flow(self):
self.client.post("/cart/add", json={"item_id": 42}, name="checkout: add_item")
self.client.post("/cart/checkout", name="checkout: submit")
self.client.get("/order/confirmation", name="checkout: confirmation")Это не даёт суммарного времени транзакции, но позволяет отслеживать каждый шаг под понятными именами.
Базовый URL задаётся через атрибут host класса или через --host в CLI. Атрибут класса полезен для тестов, которые всегда направлены на конкретный стенд:
class StagingUser(HttpUser):
host = "https://staging.example.com"
wait_time = between(1, 3)
@task
def browse(self):
self.client.get("/")Если host задан и в классе, и через CLI, значение из CLI имеет приоритет.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.