Нагрузочное тестирование с Locust

Что такое Locust? #

Locust — это open-source инструмент нагрузочного тестирования, написанный на Python. Его ключевая особенность — тесты пишутся как обычный Python-код, где поведение пользователей определяется как Python-классы. Если вы или ваша команда владеете Python, Locust предлагает самый низкий порог входа среди всех инструментов нагрузочного тестирования.

Locust использует событийно-ориентированную архитектуру (на базе gevent) вместо потоков, что позволяет одному процессу симулировать тысячи конкурентных пользователей. Инструмент включает встроенный веб-интерфейс для мониторинга тестов в реальном времени и поддерживает распределённое тестирование на нескольких машинах.

Название «Locust» (саранча) происходит от роевого поведения саранчи — вы определяете поведение пользователей, а Locust обрушивает их рой на ваше приложение.

Когда выбирать Locust #

Выбирайте Locust, когда: Команда знает Python, нужен веб-интерфейс в реальном времени, нужно использовать Python-библиотеки в тестах (драйверы БД, ML-библиотеки, кастомные протоколы) или требуется высокая гибкость поведения пользователей.

Установка #

pip install locust

Проверка:

locust --version

Ваш первый тест на Locust #

Создайте файл locustfile.py:

from locust import HttpUser, task, between

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)  # ожидание 1-3 секунды между задачами

    @task(3)
    def view_products(self):
        self.client.get("/api/products")

    @task(1)
    def view_product_detail(self):
        self.client.get("/api/products/1")

Запуск:

locust -f locustfile.py --host=https://api.example.com

Откройте http://localhost:8089 в браузере для просмотра веб-интерфейса. Введите количество пользователей, скорость запуска и нажмите Start.

Ключевые концепции #

HttpUser: Класс, представляющий виртуального пользователя. Каждый экземпляр симулирует одного пользователя.

wait_time: Управляет паузой между задачами. between(1, 3) означает случайное ожидание 1-3 секунды. Другие варианты:

• constant(2) — всегда ждать 2 секунды
• constant_pacing(5) — обеспечить, чтобы каждый цикл задач занимал ровно 5 секунд

Декоратор @task: Помечает методы как задачи пользователя. Числовой аргумент задаёт вес — @task(3) означает, что задача выполняется в 3 раза чаще, чем @task(1).

Веса задач и последовательные задачи #

Взвешенные задачи #

Веса задач моделируют реалистичное поведение пользователей. В типичном интернет-магазине просмотр происходит гораздо чаще, чем покупка:

class EcommerceUser(HttpUser):
    wait_time = between(1, 5)

    @task(10)
    def browse_products(self):
        self.client.get("/api/products")

    @task(5)
    def search(self):
        self.client.get("/api/search?q=laptop")

    @task(3)
    def view_product(self):
        self.client.get("/api/products/42")

    @task(1)
    def add_to_cart(self):
        self.client.post("/api/cart", json={"product_id": 42, "qty": 1})

Здесь просмотр происходит в 10 раз чаще, чем добавление в корзину — это отражает реальное поведение пользователей.

Последовательные задачи (TaskSets) #

Для упорядоченных сценариев используйте SequentialTaskSet:

from locust import HttpUser, SequentialTaskSet, task, between

class PurchaseFlow(SequentialTaskSet):
    @task
    def login(self):
        response = self.client.post("/api/auth/login", json={
            "username": "testuser",
            "password": "testpass"
        })
        self.token = response.json()["token"]

    @task
    def browse(self):
        self.client.get("/api/products", headers={
            "Authorization": f"Bearer {self.token}"
        })

    @task
    def add_to_cart(self):
        self.client.post("/api/cart", json={"product_id": 1, "qty": 1},
                         headers={"Authorization": f"Bearer {self.token}"})

    @task
    def checkout(self):
        self.client.post("/api/checkout",
                         headers={"Authorization": f"Bearer {self.token}"})
        self.interrupt()  # вернуться к родительскому классу

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)
    tasks = [PurchaseFlow]

Хуки жизненного цикла #

class WebsiteUser(HttpUser):
    wait_time = between(1, 3)

    def on_start(self):
        """Вызывается при запуске пользователя. Используйте для логина/настройки."""
        response = self.client.post("/api/auth/login", json={
            "username": "user1", "password": "pass123"
        })
        self.token = response.json()["token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}

    def on_stop(self):
        """Вызывается при остановке пользователя. Используйте для очистки."""
        self.client.post("/api/auth/logout", headers=self.headers)

    @task
    def browse(self):
        self.client.get("/api/products", headers=self.headers)

Пользовательская валидация #

@task
def get_products(self):
    with self.client.get("/api/products", catch_response=True) as response:
        if response.status_code != 200:
            response.failure(f"Получен статус {response.status_code}")
        elif "products" not in response.json():
            response.failure("В ответе отсутствует поле 'products'")
        elif len(response.json()["products"]) == 0:
            response.failure("Пустой список товаров")
        else:
            response.success()

Распределённое тестирование #

Locust поддерживает распределённое тестирование с архитектурой master/worker:

# Запуск master
locust -f locustfile.py --master --host=https://api.example.com

# Запуск workers (на той же или других машинах)
locust -f locustfile.py --worker --master-host=192.168.1.100
locust -f locustfile.py --worker --master-host=192.168.1.100

Master координирует тест и агрегирует результаты. Workers генерируют реальную нагрузку. Каждый worker может симулировать тысячи пользователей.

Веб-интерфейс #

Веб-интерфейс Locust на http://localhost:8089 предоставляет:

• Графики в реальном времени: Запросы в секунду, время ответа, число пользователей
• Таблица статистики: Метрики по запросам (медиана, p95, p99, max, процент ошибок)
• Вкладка ошибок: Детальные сообщения об ошибках
• Скачивание данных: Экспорт результатов в CSV
• Остановка/Сброс: Управление тестом без перезапуска

Для безголового запуска (CI/CD):

locust -f locustfile.py --headless -u 100 -r 10 --run-time 5m --host=https://api.example.com

Упражнение: Многопрофильный нагрузочный тест с Locust #

Напишите тест Locust, имитирующий три различных типа пользователей для контентной платформы.

Сценарий #

Контентная платформа имеет три типа пользователей:

1. Читатели (70%) — просматривают статьи, читают контент
2. Авторы (20%) — создают и редактируют статьи
3. Администраторы (5%) — управляют пользователями и просматривают аналитику

Требования #

1. Создайте отдельные классы пользователей для каждого типа с соответствующими весами задач
2. Используйте on_start для аутентификации
3. Добавьте пользовательскую валидацию содержимого ответов
4. Используйте between() для реалистичного think time
5. Сделайте так, чтобы класс пользователя-администратора запускался реже (используйте weight на классе)

class ReaderUser(HttpUser):
    weight = 70  # 70% пользователей — читатели
    wait_time = between(2, 5)

class AuthorUser(HttpUser):
    weight = 20  # 20% — авторы
    wait_time = between(3, 8)

class AdminUser(HttpUser):
    weight = 5   # 5% — администраторы
    wait_time = between(5, 10)

from locust import HttpUser, task, between
import random

class ReaderUser(HttpUser):
    weight = 70
    wait_time = between(2, 5)

    def on_start(self):
        response = self.client.post("/api/auth/login", json={
            "username": f"reader_{random.randint(1, 1000)}",
            "password": "readerpass"
        })
        if response.status_code == 200:
            self.token = response.json()["token"]
            self.headers = {"Authorization": f"Bearer {self.token}"}
        else:
            self.headers = {}

    @task(5)
    def browse_articles(self):
        with self.client.get("/api/articles", headers=self.headers,
                             catch_response=True) as response:
            if response.status_code == 200:
                articles = response.json().get("articles", [])
                if len(articles) > 0:
                    response.success()
                else:
                    response.failure("Статьи не возвращены")
            else:
                response.failure(f"Статус: {response.status_code}")

    @task(3)
    def read_article(self):
        article_id = random.randint(1, 100)
        with self.client.get(f"/api/articles/{article_id}",
                             headers=self.headers,
                             catch_response=True) as response:
            if response.status_code == 200:
                body = response.json()
                if "title" in body and "content" in body:
                    response.success()
                else:
                    response.failure("Статья без заголовка или содержимого")

    @task(1)
    def search_articles(self):
        queries = ["python", "testing", "qa", "automation", "ci/cd"]
        query = random.choice(queries)
        self.client.get(f"/api/search?q={query}", headers=self.headers)


class AuthorUser(HttpUser):
    weight = 20
    wait_time = between(3, 8)

    def on_start(self):
        response = self.client.post("/api/auth/login", json={
            "username": f"author_{random.randint(1, 50)}",
            "password": "authorpass"
        })
        self.token = response.json()["token"]
        self.headers = {
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        }

    @task(3)
    def view_my_articles(self):
        self.client.get("/api/articles/mine", headers=self.headers)

    @task(2)
    def create_draft(self):
        self.client.post("/api/articles", headers=self.headers, json={
            "title": f"Тестовая статья {random.randint(1, 10000)}",
            "content": "Содержимое тестовой статьи для нагрузочного тестирования.",
            "status": "draft"
        })

    @task(1)
    def edit_article(self):
        article_id = random.randint(1, 50)
        self.client.put(f"/api/articles/{article_id}", headers=self.headers, json={
            "title": "Обновлённый заголовок",
            "content": "Обновлённое содержимое."
        })


class AdminUser(HttpUser):
    weight = 5
    wait_time = between(5, 10)

    def on_start(self):
        response = self.client.post("/api/auth/login", json={
            "username": "admin",
            "password": "adminpass"
        })
        self.token = response.json()["token"]
        self.headers = {"Authorization": f"Bearer {self.token}"}

    @task(3)
    def view_analytics(self):
        self.client.get("/api/admin/analytics", headers=self.headers)

    @task(2)
    def list_users(self):
        self.client.get("/api/admin/users", headers=self.headers)

    @task(1)
    def view_system_health(self):
        self.client.get("/api/admin/health", headers=self.headers)

# С веб-интерфейсом
locust -f locustfile.py --host=https://content-api.example.com

# Безголовый режим для CI/CD
locust -f locustfile.py --headless -u 200 -r 20 --run-time 10m \
  --host=https://content-api.example.com --csv=results

Профессиональные советы #

• Custom Load Shapes: Создайте класс, наследующий LoadTestShape, для определения сложных паттернов нагрузки (spike, step, wave) с полной гибкостью Python. Это мощнее встроенного формирования нагрузки любого другого инструмента.
• Система событий: Используйте хуки событий Locust (@events.test_start.add_listener, @events.request.add_listener) для добавления кастомного логирования, метрик или уведомлений во время выполнения.
• FastHttpUser: Для максимальной пропускной способности используйте FastHttpUser вместо HttpUser. Он использует HTTP-клиент на C (geventhttpclient), который в 5-6 раз быстрее для простых запросов.
• Фильтрация по тегам: Используйте декоратор @tag('smoke') и запускайте с --tags smoke для выполнения только помеченных задач — полезно для запуска подмножеств большой тестовой сюиты.
• Docker Compose для распределённых тестов: Используйте Docker Compose для запуска master и нескольких workers одной командой, делая распределённое тестирование воспроизводимым и легко масштабируемым.