HttpUser: задачи, веса, паузы, валидация ответов

HttpUser — основной строительный блок Locust. Понимание его API позволяет описать любой реалистичный сценарий: от простого GET до авторизованных POST-запросов с проверкой результата.

#1. Атрибут self.client

Каждый экземпляр 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-сессией.

#2. Метод on_start и on_stop

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 каждого пользователя, поэтому несколько пользователей логинятся параллельно. Это именно то поведение, которое нужно имитировать.

#3. Варианты wait_time

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 задачи в секунду на пользователя

#4. Именование запросов

По умолчанию 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 в статистике.

#5. Валидация ответов: catch_response

По умолчанию 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-статус по умолчанию.

#6. Группировка нескольких запросов как одной транзакции

Иногда нужно измерить время не одного запроса, а целого пользовательского сценария (например, «добавить товар в корзину и оформить заказ»). Для этого используется менеджер событий вручную или 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")

Это не даёт суммарного времени транзакции, но позволяет отслеживать каждый шаг под понятными именами.

#7. Переменные хоста и среды

Базовый 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 имеет приоритет.