Custom Shapes

Custom Shapes allow you to have very tight programmable control over the shape of user load in your tests. This is an advanced feature, and we recommend the default for most use cases.

For our examples we will be using a very simple test that just gets the index page, but with multiple different custom shape examples.

Basic Custom Shape

This shape class will increase user count in blocks of 100 and then stop the load test after 10 minutes:

from locust import HttpUser, TaskSet, task, between
from locust import LoadTestShape


class QuickstartUser(HttpUser):
    wait_time = between(5, 9)


    @task(1)
    def index_page(self):
        self.client.get("/")


class MyCustomShape(LoadTestShape):
    time_limit = 600
    spawn_rate = 20


    def tick(self):
        run_time = self.get_run_time()


        if run_time < self.time_limit:
            # User count rounded to nearest hundred.
            user_count = round(run_time, -2)
            return (user_count, spawn_rate)


        return None

Double Wave Custom Shape

This shape class will have two peaks of users, 60 users and then 35 users over a period of 300 seconds (5 minutes).

import math
from locust import HttpUser, TaskSet, task, between
from locust import LoadTestShape


class QuickstartUser(HttpUser):
    wait_time = between(5, 9)


    @task(1)
    def index_page(self):
        self.client.get("/")




class DoubleWave(LoadTestShape):
    """
    A shape to imitate some specific user behavior. In this example, midday
    and evening meal times. First peak of users appear at time_limit/3 and
    second peak appears at 2*time_limit/3
    Settings:
        min_users -- minimum users
        peak_one_users -- users in first peak
        peak_two_users -- users in second peak
        time_limit -- total length of test
    """


    min_users = 20
    peak_one_users = 60
    peak_two_users = 40
    time_limit = 300


    def tick(self):
        run_time = round(self.get_run_time())


        if run_time < self.time_limit:
            user_count = (
                (self.peak_one_users - self.min_users)
                * math.e ** -(((run_time / (self.time_limit / 10 * 2 / 3)) - 5) ** 2)
                + (self.peak_two_users - self.min_users)
                * math.e ** -(((run_time / (self.time_limit / 10 * 2 / 3)) - 10) ** 2)
                + self.min_users
            )
            return (round(user_count), round(user_count))
        else:
            return None

Stages Custom Shape

This custom shape allows you to specify the users that are active at points throughout the test.

from locust import HttpUser, TaskSet, task, between
from locust import LoadTestShape


class QuickstartUser(HttpUser):
    wait_time = between(5, 9)


    @task(1)
    def index_page(self):
        self.client.get("/")


class StagesShape(LoadTestShape):
    """
    A simply load test shape class that has different user and spawn_rate at
    different stages.
    Keyword arguments:
        stages -- A list of dicts, each representing a stage with the following:
            duration -- When this many seconds until advancing to the next stage
            users -- Total user count
            spawn_rate -- Number of users to start/stop per second
            stop -- A boolean that can stop that test at a specific stage
        stop_at_end -- Can be set to stop once all stages have run.
    """


    stages = [
        {"duration": 60, "users": 10, "spawn_rate": 10},
        {"duration": 100, "users": 50, "spawn_rate": 10},
        {"duration": 180, "users": 100, "spawn_rate": 10},
        {"duration": 220, "users": 30, "spawn_rate": 10},
        {"duration": 230, "users": 10, "spawn_rate": 10},
        {"duration": 240, "users": 1, "spawn_rate": 1},
    ]


    def tick(self):
        run_time = self.get_run_time()


        for stage in self.stages:
            if run_time < stage["duration"]:
                tick_data = (stage["users"], stage["spawn_rate"])
                return tick_data


        return None

User Count Shape

A step load shape that waits until the target user count has been reached before waiting on a per-step timer.

import math
from locust import HttpUser, TaskSet, task, between, constant
from locust import LoadTestShape


class QuickstartUser(HttpUser):
    wait_time = between(5, 9)


    @task(1)
    def index_page(self):
        self.client.get("/")


class StepLoadShape(LoadTestShape):
    """
    A step load shape that waits until the target user count has
    been reached before waiting on a per-step timer.
    The purpose here is to ensure that a target number of users is always reached,
    regardless of how slow the user spawn rate is. The dwell time is there to
    observe the steady state at that number of users.
    Keyword arguments:
        targets_with_times -- 2-tuples, with the desired user count first,
            and the dwell (hold) time with that user count second
    """


    targets_with_times = (Step(10, 10), Step(20, 15), Step(10, 10))


    def __init__(self, *args, **kwargs):
        self.step = 0
        self.time_active = False
        super(StepLoadShape, self).__init__(*args, **kwargs)


    def tick(self):
        if self.step >= len(self.targets_with_times):
            return None


        target = self.targets_with_times[self.step]
        users = self.get_current_user_count()


        if target.users == users:
            if not self.time_active:
                self.reset_time()
                self.time_active = True
            run_time = self.get_run_time()
            if run_time > target.dwell:
                self.step += 1
                self.time_active = False


        # Spawn rate is the second value here. It is not relevant because we are
        # rate-limited by the User init rate.  We set it arbitrarily high, which
        # means "spawn as fast as you can"
        return (target.users, 100)

Step Load Shape

A step load shape that waits until the target user count has been reached before waiting on a per-step timer.

import math
from locust import HttpUser, TaskSet, task, between, constant
from locust import LoadTestShape


class QuickstartUser(HttpUser):
    wait_time = between(5, 9)


    @task(1)
    def index_page(self):
        self.client.get("/")


class StepLoadShape(LoadTestShape):
    """
    A step load shape
    Keyword arguments:
        step_time -- Time between steps
        step_load -- User increase amount at each step
        spawn_rate -- Users to stop/start per second at every step
        time_limit -- Time limit in seconds
    """


    step_time = 30
    step_load = 10
    spawn_rate = 10
    time_limit = 600


    def tick(self):
        run_time = self.get_run_time()


        if run_time > self.time_limit:
            return None


        current_step = math.floor(run_time / self.step_time) + 1
        return (current_step * self.step_load, self.spawn_rate)