By using this web site you accept our use of cookies. More information about cookies
Accept
Infopulse - Expert Software Engineering, Infrastructure Management Services
By using this web site you accept our use of cookies. More information about cookies
Accept
Infopulse - Expert Software Engineering, Infrastructure Management Services
Infopulse - Expert Software Engineering, Infrastructure Management Services
reCAPTCHA
Send message Please fill in this quick form and we will send you a free quote shortly.
* Required fields
Your privacy is important to us. We will never share your data.
Subscribe to our updates Be among the first to get exclusive content on IT insights, innovations, and best practices.
* Required fields
Your privacy is important to us. We will never share your data.
Subscribe to our New career opportunities Please fill in this quick form to be among the first to receive our updates.
* Required fields
Your privacy is important to us. We will never share your data.
Subscribe to our updates Be among the first to get exclusive content on IT insights, innovations, and best practices.
* Required fields
Your privacy is important to us. We will never share your data.
Photo of Oleksii Ostapov Send an email to Oleksii Ostapov Please fill in this quick form to contact our expert directly.
* Required fields
Your privacy is important to us. We will never share your data.
Infopulse - Expert Software Engineering, Infrastructure Management Services
Read the Full Case Study Don't miss the most interesting part of the story!
Submit this quick form to see the rest and to freely access all case studies on our website.
* Required fields
Your privacy is important to us. We will never share your data.

Performance Testing with Locust [Part 2]

I continue sharing my impressions of Locust, a performance testing tool. This article will be useful for people who liked my previous article.

In this article, I try to illustrate the benefits of writing a load test with Python code, which conveniently allows both preparing any data for the test and handling results.

Server response handling

Sometimes in performance testing it is not enough to simply receive 200 OK from HTTP server, and it is necessary to check the contents of the answer to make sure that under the load the server outputs correct data or performs correct calculations. Specifically for these cases, there is a possibility to configure criteria of successful response in Locust. Let’s check out the following example:

from locust import HttpLocust, TaskSet, task
import random as rnd
class UserBehavior(TaskSet):
   @task(1)
   def check_albums(self):
       photo_id = rnd.randint(1, 5000)
       with self.client.get(f'/photos/{photo_id}', catch_response=True, name='/photos/[id]') as response:
           if response.status_code == 200:
               album_id = response.json().get('albumId')
               if album_id % 10 != 0:
                   response.success()
               else:
                   response.failure(f'album id cannot be {album_id}')
           else:
               response.failure(f'status code is {response.status_code}')


class WebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

The example above has a single request aimed at creating load according to the following scenario:

The photos objects are requested from the server with random id in the interval between 1 and 5000. The id of the album in these objects is checked, presuming that it cannot be divisible by 10.

Several explanations seem appropriate here:

  • construction with request() as response: can be replaced with response = request() to work with a response object.
  • URL is formed according to string format syntax, this feature has been added to python 3.6, — f’/photos/{photo_id}’. This construction doesn’t exist in previous versions!
  • new argument catch_response=True, indicates to Locust that we ourselves will specify a successful server response. If it is not specified, we will still receive the answer object and will be able to process its data, but will not be able to predetermine the test result. A detailed example is provided further.
  • one more argument, name=’/photos/[id]’, is necessary to group requests in statistics. Any text can be used as a name, and we don’t have to repeat the url. Without it, each request with unique address or parameters will be recorded as a separate statistic record. It works in the following way:

Performance Testing with Locust [Part 2] - Infopulse -1

By using this argument, it is possible to perform another trick — sometimes one service with different parameters (for example with different POST requests content) executes different logic. In order for test results not to get mixed, it is possible to write several tasks, specifying a separate argument name for each.

Then we perform checks. I have done two of them. Initially I have checked whether the server returns us the answer: if response.status_code == 200:

If the answer is correct, then I check whether the album id can be divided by 10. If it is not divisible, this answer is marked as successful: response.success().

In other cases I have paid attention to the reason of response failure: response.failure(‘error text’). The following text is displayed on Failures page in the course of test execution.

Performance Testing with Locust [Part 2] - Infopulse - 2

Attentive readers could notice the absence of the exceptions handler (Exceptions), which is typical for code that is working with network interfaces. In fact, in case of timeout, connection error, and other unexpected exceptions, Locust processes them by itself and returns a response object in any case, setting the response status code to 0.

If the code generates Exceptions, it is recorded in the Exceptions tab during execution so that we can check it. The most typical situation is when json’s answer does not return the expected value, but we are already performing / we have already performed operations on it.

Before moving on I’d like to point out thatI use json server to illustrate things in the example, because it is easier to handle responses in this way. Nevertheless, in the same way it is possible to work with HTML, XML, FormData, attached files, and other data utilized by protocols based on HTTP.

Working with complicated scenarios

Almost every time when a Web-application is to undergo load testing, it quickly becomes clear that it is impossible to thoroughly cover everything using GET services only, which simply return the data.

A classic example: to test an Internet-shop it is desirable for a user to

  1. open the main page of the shop,
  2. search for goods,
  3. open the details of a product,
  4. add a product to the cart, and
  5. pay.

It is clear from the example that calling services in random order isn’t possible, and can be done only in a sequence. Moreover, the goods, the cart, and the payment method can all have unique identifiers for each user.

Using the previous example, small updates can help us conduct testing of such scenario easily. Adapting the example to our testing server:

  1. A user writes a new post.
  2. A user writes a comment to the new post.
  3. A user reads the comment
from locust import HttpLocust, TaskSet, task

class FlowException(Exception):
   pass

class UserBehavior(TaskSet):
   @task(1)
   def check_flow(self):
       # step 1
       new_post = {'userId': 1, 'title': 'my shiny new post', 'body': 'hello everybody'}
       post_response = self.client.post('/posts', json=new_post)
       if post_response.status_code != 201:
           raise FlowException('post not created')
       post_id = post_response.json().get('id')

       # step 2
       new_comment = {
           "postId": post_id,
           "name": "my comment",
           "email": "test@user.habr",
           "body": "Author is cool. Some text. Hello world!"
       }
       comment_response = self.client.post('/comments', json=new_comment)
       if comment_response.status_code != 201:
           raise FlowException('comment not created')
       comment_id = comment_response.json().get('id')

       # step 3
       self.client.get(f'/comments/{comment_id}', name='/comments/[id]')
       if comment_response.status_code != 200:
           raise FlowException('comment not read')


class WebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

I have added a new class FlowException in this example. After each step, if it is executed in an unexpected way, I run this exception class to terminate the scenario — if it is impossible to create a post, there is nothing to comment on, etc. The construction could be replaced by an usual return, but in this case in the course of execution and results analysis it will not be clearly seen in the Exceptions tab where exactly the performed scenario has failed. This is also the reason why I don’t use the try… except construction.

Making the load realistic

One can argue that in the shop case example above, all things are really linear, but the example with posts and comments is too far-fetched — posts are read at least 10 times as often as they are created. That is a reasonable observation, so let us bring the example closer to real life. There are at least two approaches:

  1. “Hardcoding” the list of posts read by users and simplifying the text code if that is possible and if backend functionality isn’t dependent on specific posts.
  2. Saving created posts and reading them if it is impossible to specify the list of posts, or if making the load realistic critically depends on what posts are read and what posts are not (I have removed commenting from the example to make the code smaller and clearer)
from locust import HttpLocust, TaskSet, task
import random as r

class UserBehavior(TaskSet):
   created_posts = []

   @task(1)
   def create_post(self):
       new_post = {'userId': 1, 'title': 'my shiny new post', 'body': 'hello everybody'}
       post_response = self.client.post('/posts', json=new_post)
       if post_response.status_code != 201:
           return
       post_id = post_response.json().get('id')
       self.created_posts.append(post_id)

   @task(10)
   def read_post(self):
       if len(self.created_posts) == 0:
           return
       post_id = r.choice(self.created_posts)
       self.client.get(f'/posts/{post_id}', name='read post')


class WebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

I have created created_posts list in UserBehavior class. Please note that this is an object and it is created not in __init__() class constructor. Therefore, unlike user’s sessions, this list is common for all users. The first task creates a post and records its id in the list. The second is 10 times as frequent, and it reads one randomly selected post from the list. An additional condition for the second task is checking if some posts have been created.

If we need each user to operate their own data, it is possible to define them in constructor in the following way:

class UserBehavior(TaskSet):
   def __init__(self, parent):
       super(UserBehavior, self).__init__(parent)
       self.created_posts = list()

Additional functionality

To launch tasks consequently, official documentation suggests using tasks annotation @seq_task(1), specifying the order number of the task in the argument

class MyTaskSequence(TaskSequence):
    @seq_task(1)
    def first_task(self):
        pass

    @seq_task(2)
    def second_task(self):
        pass

    @seq_task(3)
    @task(10)
    def third_task(self):
        pass

In the example above, each user executes first_task first, then executes second_task, and after that third_task 10 times.

To be honest, I quite like having this feature, but unlike previous examples, it is not clear how to transfer the results of the first task to the second task if necessary.

Another feature can be used for very complicated scenarios. It allows creating embedded tasks sets – creating several TaskSet classes and connecting them.

from locust import HttpLocust, TaskSet, task

class Todo(TaskSet):
   @task(3)
   def index(self):
       self.client.get("/todos")

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


class UserBehavior(TaskSet):
   tasks = {Todo: 1}

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

   @task(2)
   def posts(self):
       self.client.get("/posts")


class WebsiteUser(HttpLocust):
   task_set = UserBehavior
   min_wait = 1000
   max_wait = 2000

In the example above, Todo scenario will be launched with the probability of 1 to 6, and it will be executed until it is interrupted by UserBehavior scenario with the probability of 1 to 4. self.interrupt() is very important here because if it is absent, testing will be stuck on the subtask.

Thank you for your attention. In my last article on this topic, I will focus on distributed testing and testing without UI. It also discusses the difficulties I have encountered while testing with Locust and how they can be overcome. Stay tuned!