How to stress test a flask app with multiple image files

Viewed 210

I have a flask app which performs object detection. I'm able to get the desired output via curl by using this:

curl -X POST -F images=@images/dog.jpeg http://localhost:8000/detections

I wish to stress test it using locust. My current approach is something like this:

class UserBehavior(TaskSet):
    def _get_image_part(self, file_path, file_content_type='image/jpeg'):
        file_name = os.path.basename(file_path)
        file_content = open(file_path, 'rb')
        return file_name, file_content, file_content_type

    @task
    def detections(self):
        url = '/detections'
        
        files = {
            "images": self._get_image_part("images/dog.jpeg"),
        }
        response = self.client.post(url, files=files, verify=False)
        print(response.text)

class MyLocust(HttpUser):
    tasks = [UserBehavior]
    wait_time = between(0.1, 0.2)

How do I load a list (array) of test images for the same, so that with every request made, a randomly selected image from a directory is uploaded?

Additionally, the above snippet works fine when a small load is triggered (2-3 users with <5rps) but on higher requests or concurrent users, the application throws error (the most common one being FileNotFoundError). Is it justified?

Thanks in advance

2 Answers

Load the images into a list (as a global or class level variable), then just pick one randomly in the task.

What you want to avoid is reloading the image from disk every time, because there is really no need.

Based on Cyberwiz's answer, I solved it by creating a list of images in a directory and then randomly selecting an image from this list for every request.

Here's my implementation:

def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.image_list = os.listdir('images')

@task
def detections(self):
    url = '/detections'
    image_name = random.choice(self.image_list)
    
    files = {
        "images": self._get_image_part(os.path.join(os.getcwd(), 'images', image_name))
    }
    response = self.client.post(url, files=files, verify=False)
    print(response.text)
Related