Locust with Python: Introduction, Correlating Variables, and Basic Assertions

In this post, we will show you how to start your first test using Locust and Python. Plus, get a step-by-step overview of how to correlate valuables and assert your script.

What is Locust Python?

Locust is a great open source load testing tool for developers written in Python, since tests can be created as code. It helps teams find bottlenecks and other performance issues.

How to Use Python + Locust

Getting Started: Installing Python

In order to run Locust and Python, you need to have Python installed. (You'll need to download Python first.) Then, all you have to do is run the following command:

pip3 install locust

Scripting and Executing the Load Test from the Locust GUI

The next step is creating a file for our script named locustfile.py. In this file we will define the HTTP requests to be executed in our load test. Using this file name for the script enables Locust to automatically find the file. (If you want to use another name for the file, you’ll need to add the parameter -f and the file name when executing. I will show you how below).

In this example we will load test the site https://www.demoblaze.com/.

1. Creating a Basic Script

To start we will just write a script that makes a call to the demoblaze main page as shown below:

from locust import HttpUser, task
            
class User(HttpUser):
    @task
    def mainPage(self):
        self.client.get("/")

Notice that the url of the site under test is not specified in the script. Rather, it is specified from the UI when running the test.

Locust will only run functions with the decorator @task, so we must remember to add it. For cases when there are multiple tasks defined, they’ll execute randomly by default.

2. Running the Script

In order to run this test we’ll need to execute the command locust in the script’s directory from the command line, which will start a web user interface on port 8089. Just navigate to http://localhost:8089 on the browser to access it.

Note: if you are getting an error as the port is being used by another program, you can change it using the command locust --web-port [port]. And as I mentioned above, if you want to run a script with a different name, you can execute locust -f [file name].

The number of users, users started per second and host (the URL under test) can be chosen from Locust’s web user interface. Locust will create the full url for each request using the paths from the script and the host specified here.

Now you can run the test by clicking the Start swarming button and the browser will show a dashboard like the following:

In this dashboard we’ll see a report including the number of requests executed, number of failed requests, 90 percentile and other statistics in real time. By default, the virtual users will keep running until the test is stopped.

If you want to run the script without using the web UI, you can run the command:

locust --headless -u 1 -r 1 -H https://www.demoblaze.com

Where -u specifies the number of users, -r the spawn rate, and -H the host.

3. Creating a More Advanced Locust Python Script

Now that we understand how to create and run a basic test, let’s do one with a more complex workflow.

This workflow consists of 4 steps:

1. Enter the website 
2. Log in
3. Add a product to the cart
4. Enter the cart

Once we have obtained the HTTP requests for each step of the workflow, we can create a method for each user action.

from locust import HttpUser, SequentialTaskSet, task, between
            
class User(HttpUser):    
    @task
    class SequenceOfTasks(SequentialTaskSet):
        wait_time = between(1, 5)
        @task
        def mainPage(self):
            self.client.get("/")
            self.client.get("https://api.demoblaze.com/entries")
        @task
        def login(self):
            self.client.options("https://api.demoblaze.com/login")
            self.client.post("https://api.demoblaze.com/login",json={"username":"aaaa","password":"YWFhYQ=="})
            self.client.options("https://api.demoblaze.com/check")
            self.client.get("https://api.demoblaze.com/entries")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})            
        @task
        def clickProduct(self):
            self.client.get("/prod.html?idp_=1")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})
        @task
        def addToCart(self):
            self.client.options("https://api.demoblaze.com/addtocart")
            self.client.post("https://api.demoblaze.com/addtocart",json={"id":"fb3d5d23-f88c-80d9-a8de-32f1b6034bfd","cookie":"YWFhYTE2MzA5NDU=","prod_id":1,"flag":'true'})
        @task 
        def viewCart(self):
            self.client.get("/cart.html")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/viewcart")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})
            self.client.post("https://api.demoblaze.com/viewcart",json={"cookie":"YWFhYTE2MzA5NDU=","flag":'true'})
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":"YWFhYTE2MzA5NDU="})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})

As you can see, all the methods are included in a class using SequentialTaskSet, so they can be executed sequentially in the same order that they were declared.

In addition, by using wait_time we can add a pause in between tasks. In this case it’s a random pause between 1 and 5 seconds.

You’ll also notice that the requests to the API are written using the full url, as the Locust web user interface only allows using one URL in the host field.

Correlating Variables

The next step is to correlate the hardcoded token. It’s important to correlate parameters that are dynamic, and we know that this token changes every time that the user logs in. The token could expire and not having it parametrized would make the script stop working.

Analyzing the HTTP requests workflow, we can see that the token sent in the /check post can be extracted from the /login response. We can then save it into a variable and use it in all the following requests that need it.

The response has the token in this format:

"Auth_token: YWFhYTE2MzA1ODg="

So the following regex expression can be used to extract it:

"Auth_token: (.+?)"

Next, the module re needs to be imported and the match method is used to save the extracted value in a variable.

So to extract the token from the response, we will first save the response to a variable like this:

  response = self.client.post("https://api.demoblaze.com/login",json={"username":"aaaa","password":"YWFhYQ=="})        

Now we can define a global variable and extract the token using the regular expression.

global token 
  token = re.match("\"Auth_token: (.+?)\"",response.text)[1]

And we can use the variable in the following requests

self.client.post("https://api.demoblaze.com/check",json={"token":token}

This is what the login and click product transactions look like:

 @task
        def login(self):
            self.client.options("https://api.demoblaze.com/login")
            response = self.client.post("https://api.demoblaze.com/login",json={"username":"aaaa","password":"YWFhYQ=="})
            global token 
            token = re.match("\"Auth_token: (.+?)\"",response.text)[1]
            self.client.options("https://api.demoblaze.com/check")
            self.client.get("https://api.demoblaze.com/entries")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})            
        @task
        def clickProduct(self):
            self.client.get("/prod.html?idp_=1")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})

Note: In this example we used only one user. You can also learn how to run the script with multiple users.

Using Assertions with Locust & Python

Now I will show you how to add a simple assertion to validate that the product added to the cart has been added correctly.

Locust does not have many built-in functionalities, but using Python you can easily add on custom functionalities.

response.failure(“Error message”) can be used to mark a request as failed. In order to get the assertion working properly, this function should be used inside an if clause and the catch_response argument should be added to validate the response as shown below.

 @task 
        def viewCart(self):
            self.client.get("/cart.html")
            self.client.options("https://api.demoblaze.com/check")
            self.client.options("https://api.demoblaze.com/viewcart")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})
            with self.client.post("https://api.demoblaze.com/viewcart",catch_response=True,json={"cookie":token,"flag":'true'}) as response:
                if '"prod_id":1' not in response.text:
                    response.failure("Assert failure, response does not contain expected prod_id")
            self.client.options("https://api.demoblaze.com/view")
            self.client.post("https://api.demoblaze.com/check",json={"token":token})
            self.client.post("https://api.demoblaze.com/view",json={"id":"1"})

If we run the test we can see that it has no errors. But how can we know that the assertion is working properly? Let’s change the product id to one that doesn’t exist in the response.

  with self.client.post("https://api.demoblaze.com/viewcart",catch_response=True,json={"cookie":token,"flag":'true'}) as response:
                if '"prod_id":1234' not in response.text:
                    response.failure("Assert failure, response does not contain expected prod_id")

When the test is executed again, the /viewcart POST will fail and the error message defined can be seen in the Failures tab.

If you are going to use multiple assertions, it might be better to create a function you can avoid rewriting similar code, like this one:

def assertContains(response,text):
    with response as r:
        if text not in r.text:
            r.failure("Expected "+ response.text + " to contain "+ text)

and you can call it like is shown below every time you want to make an assertion with text:

            assertContains(self.client.post("https://api.demoblaze.com/viewcart",catch_response=True,json={"cookie":token,"flag":'true'}),'"prod_id":1')

Once you finish your Locust script, you can run it in BlazeMeter to scale, integrate in CI/CD and see advanced reporting. 

START TESTING NOW

Related Resources

• Writing Python API Tests with Apiritif
• Using JMeter Variables With Multiple Thread Groups