TaskSet и SequentialTaskSet: сложные сценарии

TaskSet позволяет организовать задачи в логические группы, имитируя поведение пользователя внутри определённой части приложения — например, раздела каталога или личного кабинета.

#1. Зачем нужен TaskSet

Простого HttpUser с несколькими @task-методами достаточно для базовых сценариев. Но реальное поведение пользователя часто иерархическое: пользователь заходит в каталог, просматривает несколько страниц товаров, затем переходит в корзину и совершает ряд действий там.

TaskSet позволяет описать такую иерархию: HttpUser определяет «верхний уровень» (какой раздел посещать), а каждый TaskSet — поведение внутри раздела:

from locust import HttpUser, TaskSet, task, between

class CatalogBehavior(TaskSet):
    @task(3)
    def browse_products(self):
        self.client.get("/catalog")

    @task(1)
    def view_product(self):
        self.client.get("/catalog/product-1")

    @task(1)
    def leave_catalog(self):
        self.interrupt()  # вернуться на уровень выше

class ShopUser(HttpUser):
    tasks = [CatalogBehavior]
    wait_time = between(1, 3)

Когда Locust выбирает CatalogBehavior, пользователь «входит» в этот набор задач и выполняет их до тех пор, пока не будет вызван self.interrupt().

#2. Метод self.interrupt

self.interrupt() — единственный способ выйти из TaskSet обратно к родительскому уровню (к HttpUser или другому TaskSet). Без него пользователь будет навсегда «застрять» в данном TaskSet.

По умолчанию interrupt() выбрасывает исключение, которое перехватывает Locust и возвращает управление родителю. Аргумент reschedule=True (по умолчанию) означает, что после выхода из TaskSet пользователь сразу выбирает следующую задачу. reschedule=False добавит дополнительное ожидание wait_time:

class CheckoutBehavior(TaskSet):
    @task
    def add_to_cart(self):
        self.client.post("/cart/add", json={"item_id": 1})

    @task
    def finish_shopping(self):
        # Выходим из TaskSet, выждав wait_time перед следующей задачей
        self.interrupt(reschedule=False)

#3. SequentialTaskSet

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

from locust import HttpUser, SequentialTaskSet, task, between

class RegistrationFlow(SequentialTaskSet):
    @task
    def register(self):
        self.client.post("/auth/register", json={
            "email": "user@test.com",
            "password": "pass123"
        })

    @task
    def verify_email(self):
        self.client.post("/auth/verify", json={"code": "123456"})

    @task
    def setup_profile(self):
        self.client.post("/profile/setup", json={"name": "Test User"})
        self.interrupt()

class NewUser(HttpUser):
    tasks = [RegistrationFlow]
    wait_time = between(1, 2)

Каждый вызов @task в SequentialTaskSet увеличивает внутренний индекс. После последней задачи цикл начинается сначала (если не вызван interrupt()).

#4. Вложенные TaskSet

TaskSet могут быть вложены произвольно глубоко. Дочерний TaskSet указывается как значение в tasks родителя:

class ProductPageBehavior(TaskSet):
    @task(2)
    def view_details(self):
        self.client.get("/product/details")

    @task(1)
    def view_reviews(self):
        self.client.get("/product/reviews")

    @task(1)
    def back_to_catalog(self):
        self.interrupt()

class CatalogBehavior(TaskSet):
    tasks = {ProductPageBehavior: 2, "browse_catalog": 3}

    @task(3)
    def browse_catalog(self):
        self.client.get("/catalog")

    @task(1)
    def exit_catalog(self):
        self.interrupt()

class ShopUser(HttpUser):
    tasks = [CatalogBehavior]
    wait_time = between(0.5, 2)

Веса в словаре tasks работают так же, как в @task(N): ProductPageBehavior вызывается в 2 раза чаще относительно своего веса в словаре.

#5. Передача данных между задачами

TaskSet — обычный Python-класс, поэтому для передачи состояния между задачами используйте атрибуты экземпляра:

class CartBehavior(TaskSet):
    def on_start(self):
        self.cart_id = None

    @task
    def create_cart(self):
        response = self.client.post("/cart/create")
        self.cart_id = response.json()["id"]

    @task
    def add_item(self):
        if self.cart_id:
            self.client.post(f"/cart/{self.cart_id}/items", json={"product_id": 1})

    @task
    def checkout(self):
        if self.cart_id:
            self.client.post(f"/cart/{self.cart_id}/checkout")
            self.cart_id = None
            self.interrupt()

Важно: self.user внутри TaskSet — ссылка на экземпляр HttpUser. Через self.user можно получить данные, инициализированные в on_start пользователя.

#6. TaskSet vs SequentialTaskSet: когда что использовать

TaskSet (случайный порядок) подходит для имитации «блужданий» по приложению: пользователь может смотреть каталог, случайно переходить на страницы товаров и возвращаться назад.

SequentialTaskSet подходит для строго детерминированных сценариев: оформление заказа, многошаговая регистрация, wizard-интерфейсы. Любое нарушение порядка шагов в таких сценариях вызвало бы ошибку бизнес-логики.

Смешанный подход: SequentialTaskSet для критического пути + TaskSet для «исследовательского» поведения между шагами.