Writing a locustfile

Now, lets look at a more complete/realistic example of what your tests might look like:

import time
from locust import HttpUser, task, between

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

    @task
    def hello_world(self):
        self.client.get("/hello")
        self.client.get("/world")

    @task(3)
    def view_items(self):
        for item_id in range(10):
            self.client.get(f"/item?id={item_id}", name="/item")
            time.sleep(1)

    def on_start(self):
        self.client.post("/login", json={"username":"foo", "password":"bar"})

Let’s break it down

import time
from locust import HttpUser, task, between

A locust file is just a normal Python module, it can import code from other files or packages.

class QuickstartUser(HttpUser):

Here we define a class for the users that we will be simulating. It inherits from HttpUser which gives each user a client attribute, which is an instance of HttpSession, that can be used to make HTTP requests to the target system that we want to load test. When a test starts, locust will create an instance of this class for every user that it simulates, and each of these users will start running within their own green gevent thread.

For a file to be a valid locustfile it must contain at least one class inheriting from User.

wait_time = between(1, 5)

Our class defines a wait_time that will make the simulated users wait between 1 and 5 seconds after each task (see below) is executed. For more info see wait_time attribute.

@task
def hello_world(self):
    ...

Methods decorated with @task are the core of your locust file. For every running user, Locust creates a greenlet (micro-thread), that will call those methods.

@task
def hello_world(self):
    self.client.get("/hello")
    self.client.get("/world")

@task(3)
def view_items(self):
...

We’ve declared two tasks by decorating two methods with @task, one of which has been given a higher weight (3). When our QuickstartUser runs it’ll pick one of the declared tasks - in this case either hello_world or view_items - and execute it. Tasks are picked at random, but you can give them different weighting. The above configuration will make Locust three times more likely to pick view_items than hello_world. When a task has finished executing, the User will then sleep during its wait time (in this case between 1 and 5 seconds). After its wait time it’ll pick a new task and keep repeating that.

Note that only methods decorated with @task will be picked, so you can define your own internal helper methods any way you like.

self.client.get("/hello")

The self.client attribute makes it possible to make HTTP calls that will be logged by Locust. For information on how to make other kinds of requests, validate the response, etc, see Using the HTTP Client.

Note

HttpUser is not a real browser, and thus will not parse an HTML response to load resources or render the page. It will keep track of cookies though.

@task(3)
def view_items(self):
    for item_id in range(10):
        self.client.get(f"/item?id={item_id}", name="/item")
        time.sleep(1)

In the view_items task we load 10 different URLs by using a variable query parameter. In order to not get 10 separate entries in Locust’s statistics - since the stats is grouped on the URL - we use the name parameter to group all those requests under an entry named "/item" instead.

def on_start(self):
    self.client.post("/login", json={"username":"foo", "password":"bar"})

Additionally we’ve declared an on_start method. A method with this name will be called for each simulated user when they start. For more info see on_start and on_stop methods.

Auto-generating a locustfile

You can use har2locust to generate locustfiles based on a browser recording (HAR-file).

It is particularly useful for beginners that are not used to writing their own locustfile, but also highly customizable for more advanced use cases.

Note

har2locust is still in beta. It may not always generate correct locustfiles, and its interface may change between versions.

User class

A user class represents one type of user/scenario for your system. When you do a test run you specify the number of concurrent users you want to simulate and Locust will create an instance per user. You can add any attributes you like to these classes/instances, but there are some that have special meaning to Locust:

wait_time attribute

A User’s wait_time method makes it easy to introduce delays after each task execution. If no wait_time is specified, the next task will be executed as soon as one finishes.

  • constant for a fixed amount of time

  • between for a random time between a min and max value

For example, to make each user wait between 0.5 and 10 seconds between every task execution:

from locust import User, task, between

class MyUser(User):
    @task
    def my_task(self):
        print("executing my_task")

    wait_time = between(0.5, 10)
  • constant_throughput for an adaptive time that ensures the task runs (at most) X times per second.

  • constant_pacing for an adaptive time that ensures the task runs (at most) once every X seconds (it is the mathematical inverse of constant_throughput).

Note

For example, if you want Locust to run 500 task iterations per second at peak load, you could use wait_time = constant_throughput(0.1) and a user count of 5000.

Wait time can only constrain the throughput, not launch new Users to reach the target. So, in our example, the throughput will be less than 500 if the time for the task iteration exceeds 10 seconds.

Wait time is applied after task execution, so if you have a high spawn rate/ramp up you may end up exceeding your target during ramp-up.

Wait times apply to tasks, not requests. For example, if you specify wait_time = constant_throughput(2) and do two requests in your tasks, your request rate/RPS will be 4 per User.

It’s also possible to declare your own wait_time method directly on your class. For example, the following User class would sleep for one second, then two, then three, etc.

class MyUser(User):
    last_wait_time = 0

    def wait_time(self):
        self.last_wait_time += 1
        return self.last_wait_time

    ...

weight and fixed_count attributes

If more than one user class exists in the file, and no user classes are specified on the command line, Locust will spawn an equal number of each of the user classes. You can also specify which of the user classes to use from the same locustfile by passing them as command line arguments:

$ locust -f locust_file.py WebUser MobileUser

If you wish to simulate more users of a certain type you can set a weight attribute on those classes. Say for example, web users are three times more likely than mobile users:

class WebUser(User):
    weight = 3
    ...

class MobileUser(User):
    weight = 1
    ...

Also you can set the fixed_count attribute. In this case the weight attribute will be ignored and the exact count users will be spawned. These users are spawned first. In the example below, only one instance of AdminUser will be spawned, to make some specific work with more accurate control of request count independently of total user count.

class AdminUser(User):
    wait_time = constant(600)
    fixed_count = 1

    @task
    def restart_app(self):
        ...

class WebUser(User):
    ...

host attribute

The host attribute is a URL prefix (e.g. https://google.com) to the host you want to test. It is automatically added to requests, so you can do self.client.get("/") for example.

You can overwrite this value in Locust’s web UI or on the command line, using the --host option.

tasks attribute

A User class can have tasks declared as methods under it using the @task decorator, but one can also specify tasks using the tasks attribute which is described in more details below.

environment attribute

A reference to the environment in which the user is running. Use this to interact with the environment, or the runner which it contains. E.g. to stop the runner from a task method:

self.environment.runner.quit()

If run on a standalone locust instance, this will stop the entire run. If run on worker node, it will stop that particular node.

on_start and on_stop methods

Users (and TaskSets) can declare an on_start method and/or on_stop method. A User will call its on_start method when it starts running, and its on_stop method when it stops running. For a TaskSet, the on_start method is called when a simulated user starts executing that TaskSet, and on_stop is called when the simulated user stops executing that TaskSet (when interrupt() is called, or the user is killed).

Tasks

When a load test is started, an instance of a User class will be created for each simulated user and they will start running within their own green thread. When these users run they pick tasks that they execute, sleep for awhile, and then pick a new task and so on.

The tasks are normal python callables and - if we were load-testing an auction website - they could do stuff like “loading the start page”, “searching for some product”, “making a bid”, etc.

@task decorator

The easiest way to add a task for a User is by using the task decorator.

from locust import User, task, constant

class MyUser(User):
    wait_time = constant(1)

    @task
    def my_task(self):
        print("User instance (%r) executing my_task" % self)

@task takes an optional weight argument that can be used to specify the task’s execution ratio. In the following example, task2 will be twice as likely to be selected as task1:

from locust import User, task, between

class MyUser(User):
    wait_time = between(5, 15)

    @task(3)
    def task1(self):
        pass

    @task(6)
    def task2(self):
        pass

tasks attribute

Another way to define the tasks of a User is by setting the tasks attribute.

The tasks attribute is either a list of Tasks, or a <Task : int> dict, where Task is either a python callable or a TaskSet class. If the task is a normal python function they receive a single argument which is the User instance that is executing the task.

Here is an example of a User task declared as a normal python function:

from locust import User, constant

def my_task(user):
    pass

class MyUser(User):
    tasks = [my_task]
    wait_time = constant(1)

If the tasks attribute is specified as a list, each time a task is to be performed, it will be randomly chosen from the tasks attribute. If however, tasks is a dict - with callables as keys and ints as values - the task that is to be executed will be chosen at random but with the int as ratio. So with a task that looks like this:

{my_task: 3, another_task: 1}

my_task would be 3 times as likely to be executed as another_task.

Internally the above dict will actually be expanded into a list (and the tasks attribute is updated) that looks like this:

[my_task, my_task, my_task, another_task]

and then Python’s random.choice() is used to pick tasks from the list.

@tag decorator

By tagging tasks using the @tag decorator, you can be picky about what tasks are executed during the test using the --tags and --exclude-tags arguments. Consider the following example:

from locust import User, constant, task, tag

class MyUser(User):
    wait_time = constant(1)

    @tag('tag1')
    @task
    def task1(self):
        pass

    @tag('tag1', 'tag2')
    @task
    def task2(self):
        pass

    @tag('tag3')
    @task
    def task3(self):
        pass

    @task
    def task4(self):
        pass

If you started this test with --tags tag1, only task1 and task2 would be executed during the test. If you started it with --tags tag2 tag3, only task2 and task3 would be executed.

--exclude-tags will behave in the exact opposite way. So, if you start the test with --exclude-tags tag3, only task1, task2, and task4 will be executed. Exclusion always wins over inclusion, so if a task has a tag you’ve included and a tag you’ve excluded, it will not be executed.

Events

If you want to run some setup code as part of your test, it is often enough to put it at the module level of your locustfile, but sometimes you need to do things at particular times in the run. For this need, Locust provides event hooks.

test_start and test_stop

If you need to run some code at the start or stop of a load test, you should use the test_start and test_stop events. You can set up listeners for these events at the module level of your locustfile:

from locust import events

@events.test_start.add_listener
def on_test_start(environment, **kwargs):
    print("A new test is starting")

@events.test_stop.add_listener
def on_test_stop(environment, **kwargs):
    print("A new test is ending")

init

The init event is triggered at the beginning of each Locust process. This is especially useful in distributed mode where each worker process (not each user) needs a chance to do some initialization. For example, let’s say you have some global state that all users spawned from this process will need:

from locust import events
from locust.runners import MasterRunner

@events.init.add_listener
def on_locust_init(environment, **kwargs):
    if isinstance(environment.runner, MasterRunner):
        print("I'm on master node")
    else:
        print("I'm on a worker or standalone node")

Other events

See extending locust using event hooks for other events and more examples of how to use them.

HttpUser class

HttpUser is the most commonly used User. It adds a client attribute which is used to make HTTP requests.

from locust import HttpUser, task, between

class MyUser(HttpUser):
    wait_time = between(5, 15)

    @task(4)
    def index(self):
        self.client.get("/")

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

client attribute / HttpSession

client is an instance of HttpSession. HttpSession is a subclass/wrapper for requests.Session, so its features are well documented and should be familiar to many. What HttpSession adds is mainly reporting of the request results into Locust (success/fail, response time, response length, name).

It contains methods for all HTTP methods: get, post, put, …

Just like requests.Session, it preserves cookies between requests so it can easily be used to log in to websites.

Make a POST request, look at the response and implicitly reuse any session cookie we got for a second request
response = self.client.post("/login", {"username":"testuser", "password":"secret"})
print("Response status code:", response.status_code)
print("Response text:", response.text)
response = self.client.get("/my-profile")

HttpSession catches any requests.RequestException thrown by Session (caused by connection errors, timeouts or similar), instead returning a dummy Response object with status_code set to 0 and content set to None.

Validating responses

Requests are considered successful if the HTTP response code is OK (<400), but it is often useful to do some additional validation of the response.

You can mark a request as failed by using the catch_response argument, a with-statement and a call to response.failure()

with self.client.get("/", catch_response=True) as response:
    if response.text != "Success":
        response.failure("Got wrong response")
    elif response.elapsed.total_seconds() > 0.5:
        response.failure("Request took too long")

You can also mark a request as successful, even if the response code was bad:

with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        response.success()

You can even avoid logging a request at all by throwing an exception and then catching it outside the with-block. Or you can throw a locust exception, like in the example below, and let Locust catch it.

from locust.exception import RescheduleTask
...
with self.client.get("/does_not_exist/", catch_response=True) as response:
    if response.status_code == 404:
        raise RescheduleTask()

REST/JSON APIs

FastHttpUser provides a ready-made rest method, but you can also do it yourself:

from json import JSONDecodeError
...
with self.client.post("/", json={"foo": 42, "bar": None}, catch_response=True) as response:
    try:
        if response.json()["greeting"] != "hello":
            response.failure("Did not get expected value in greeting")
    except JSONDecodeError:
        response.failure("Response could not be decoded as JSON")
    except KeyError:
        response.failure("Response did not contain expected key 'greeting'")

Grouping requests

It’s very common for websites to have pages whose URLs contain some kind of dynamic parameter(s). Often it makes sense to group these URLs together in User’s statistics. This can be done by passing a name argument to the HttpSession's different request methods.

Example:

# Statistics for these requests will be grouped under: /blog/?id=[id]
for i in range(10):
    self.client.get("/blog?id=%i" % i, name="/blog?id=[id]")

There may be situations where passing in a parameter into request function is not possible, such as when interacting with libraries/SDK’s that wrap a Requests session. An alternative way of grouping requests is provided by setting the client.request_name attribute.

# Statistics for these requests will be grouped under: /blog/?id=[id]
self.client.request_name="/blog?id=[id]"
for i in range(10):
    self.client.get("/blog?id=%i" % i)
self.client.request_name=None

If you want to chain multiple groupings with minimal boilerplate, you can use the client.rename_request() context manager.

@task
def multiple_groupings_example(self):
    # Statistics for these requests will be grouped under: /blog/?id=[id]
    with self.client.rename_request("/blog?id=[id]"):
        for i in range(10):
            self.client.get("/blog?id=%i" % i)

    # Statistics for these requests will be grouped under: /article/?id=[id]
    with self.client.rename_request("/article?id=[id]"):
        for i in range(10):
            self.client.get("/article?id=%i" % i)

Using catch_response and accessing request_meta directly, you can even rename requests based on something in the response.

with self.client.get("/", catch_response=True) as resp:
    resp.request_meta["name"] = resp.json()["name"]

HTTP Proxy settings

To improve performance, we configure requests to not look for HTTP proxy settings in the environment by setting requests.Session’s trust_env attribute to False. If you don’t want this you can manually set locust_instance.client.trust_env to True. For further details, refer to the documentation of requests.

Connection pooling

As every HttpUser creates new HttpSession, every user instance has its own connection pools. This is similar to how real users would interact with a web server.

However, if you want to share connections among all users, you can use a single pool manager. To do this, set pool_manager class attribute to an instance of urllib3.PoolManager.

from locust import HttpUser
from urllib3 import PoolManager

class MyUser(HttpUser):
    # All users will be limited to 10 concurrent connections at most.
    pool_manager = PoolManager(maxsize=10, block=True)

For more configuration options, refer to the urllib3 documentation.

TaskSets

TaskSets is a way to structure tests of hierarchical web sites/systems. You can read more about it here.

Examples

There are lots of locustfile examples here

How to structure your test code

It’s important to remember that the locustfile.py is just an ordinary Python module that is imported by Locust. From this module you’re free to import other python code just as you normally would in any Python program. The current working directory is automatically added to python’s sys.path, so any python file/module/packages that resides in the working directory can be imported using the python import statement.

For small tests, keeping all of the test code in a single locustfile.py should work fine, but for larger test suites, you’ll probably want to split the code into multiple files and directories.

How you structure the test source code is of course entirely up to you, but we recommend that you follow Python best practices. Here’s an example file structure of an imaginary Locust project:

  • Project root

    • common/

      • __init__.py

      • auth.py

      • config.py

    • locustfile.py

    • requirements.txt (External Python dependencies is often kept in a requirements.txt)

A project with multiple locustfiles could also keep them in a separate subdirectory:

  • Project root

    • common/

      • __init__.py

      • auth.py

      • config.py

    • my_locustfiles/

      • api.py

      • website.py

    • requirements.txt

With any of the above project structure, your locustfile can import common libraries using:

import common.auth