Кастомные клиенты: FastHttpUser, WebSocket, gRPC

Locust не ограничен HTTP. Любой протокол — WebSocket, gRPC, TCP, AMQP — можно протестировать, написав собственный клиент и правильно передавая метрики в статистику Locust.

#1. FastHttpUser

FastHttpUser — это более быстрая альтернатива стандартному HttpUser. Он использует geventhttpclient вместо requests, что даёт меньший overhead CPU на каждый запрос.

Разница существенна при большом числе пользователей: HttpUser с 2000 пользователями может нагрузить CPU генератора до предела, тогда как FastHttpUser справляется с тем же числом при значительно меньшей нагрузке:

from locust import task, between
from locust.contrib.fasthttp import FastHttpUser

class FastApiUser(FastHttpUser):
    wait_time = between(0.5, 1.5)

    @task
    def browse(self):
        self.client.get("/api/items")

    @task
    def create(self):
        self.client.post("/api/items", json={"name": "test"})

API FastHttpUser практически идентично HttpUser: те же методы (get, post, put, delete), те же параметры name, catch_response=True. Основное ограничение: не поддерживает cert и некоторые расширенные параметры requests.

#2. Разница между HttpUser и FastHttpUser

Стандартный HttpUser основан на requests.Session и хорошо подходит для сценариев, где нужна совместимость с экосистемой requests (middleware, custom adapters, сложная cookie-логика).

FastHttpUser использует geventhttpclient.HTTPClient — низкоуровневый клиент с минимальным overhead. Используйте FastHttpUser когда:

Нужно симулировать более 500–1000 пользователей на одной машине
CPU генератора нагрузки становится узким местом
Не нужны специфические возможности requests

#3. Кастомный клиент для WebSocket

Для тестирования WebSocket-сервисов нужно создать собственный клиент и вручную записывать метрики через request_meta:

import time
import gevent
from locust import User, task, between, events
from websocket import create_connection

class WebSocketClient:
    def __init__(self, host, environment):
        self.host = host
        self.environment = environment
        self.ws = create_connection(f"ws://{host}/ws")

    def send_and_receive(self, message, name="websocket"):
        start_time = time.time()
        exception = None
        response_length = 0

        try:
            self.ws.send(message)
            response = self.ws.recv()
            response_length = len(response)
        except Exception as e:
            exception = e
        finally:
            elapsed_ms = int((time.time() - start_time) * 1000)
            self.environment.events.request.fire(
                request_type="WebSocket",
                name=name,
                response_time=elapsed_ms,
                response_length=response_length,
                exception=exception,
            )

        return response

    def close(self):
        self.ws.close()


class WebSocketUser(User):
    wait_time = between(1, 3)

    def on_start(self):
        self.ws_client = WebSocketClient(self.host, self.environment)

    def on_stop(self):
        self.ws_client.close()

    @task
    def send_message(self):
        self.ws_client.send_and_receive(
            '{"type": "ping"}',
            name="ws/ping"
        )

Ключевой момент — вызов self.environment.events.request.fire(). Именно через него метрики попадают в статистику Locust и отображаются в Web UI.

#4. Параметры events.request.fire

Метод request.fire() принимает именованные аргументы:

request_type — произвольная строка для типа запроса ("GET", "WebSocket", "gRPC")
name — имя/путь запроса, под которым он появится в статистике
response_time — время в миллисекундах
response_length — размер ответа в байтах (0, если неизвестен)
exception — объект исключения если запрос упал, иначе None

При exception=None запрос считается успешным. При exception=SomeException(...) — неуспешным, с этим исключением в колонке Failures.

#5. Кастомный клиент для gRPC

gRPC-сервисы тестируются аналогично — через базовый класс User (без Http) и вызов events.request.fire():

import time
import grpc
from locust import User, task, between

# Импорт сгенерированных protobuf-классов
from myservice_pb2 import HelloRequest
from myservice_pb2_grpc import GreeterStub

class GrpcUser(User):
    wait_time = between(0.5, 2)
    abstract = True  # не создавать пользователей этого базового класса напрямую

    def on_start(self):
        self.channel = grpc.insecure_channel(f"{self.host}:50051")
        self.stub = GreeterStub(self.channel)

    def on_stop(self):
        self.channel.close()

    def grpc_call(self, stub_method, request, name):
        start = time.time()
        exception = None
        response_length = 0

        try:
            response = stub_method(request)
            response_length = response.ByteSize()
        except grpc.RpcError as e:
            exception = e

        self.environment.events.request.fire(
            request_type="gRPC",
            name=name,
            response_time=int((time.time() - start) * 1000),
            response_length=response_length,
            exception=exception,
        )


class HelloUser(GrpcUser):
    @task
    def say_hello(self):
        self.grpc_call(
            self.stub.SayHello,
            HelloRequest(name="Locust"),
            name="/helloworld.Greeter/SayHello"
        )

Атрибут abstract = True указывает Locust не создавать пользователей этого класса напрямую — он служит базовым классом для переиспользования.

#6. Класс User без HTTP

При наследовании от User (а не HttpUser) атрибут self.client недоступен. Это осознанный выбор: User — минималистичный базовый класс без предположений о протоколе. Все соединения создаются вручную в on_start.

Это подходит для тестирования:

WebSocket и Server-Sent Events
gRPC
AMQP (RabbitMQ, ActiveMQ)
Redis команд через python-redis
TCP/UDP на уровне сокетов
Любого другого протокола или бинарного API