Way to use locust.io by supplying user list

Question

I need to stress-test a system and http://locust.io seems like the best way to go about this. However, it looks like it is set up to use the same user every time. I need each spawn to log in as a different user. How do I go about setting that up? Alternatively, is there another system that would be good to use?

Answer 1

Locust author here.

By default, each HttpLocust user instance has an HTTP client that has it's own separate session.

Locust doesn't have any feature for providing a list of user credentials or similar. However, your load testing scripts are just python code, and luckily it's trivial to implement this yourself.

Here's a short example:

# locustfile.py

from locust import HttpLocust, TaskSet, task

USER_CREDENTIALS = [
    ("user1", "password"),
    ("user2", "password"),
    ("user3", "password"),
]

class UserBehaviour(TaskSet):
    def on_start(self):
        if len(USER_CREDENTIALS) > 0:
            user, passw = USER_CREDENTIALS.pop()
            self.client.post("/login", {"username":user, "password":passw})

    @task
    def some_task(self):
        # user should be logged in here (unless the USER_CREDENTIALS ran out)
        self.client.get("/protected/resource")

class User(HttpLocust):
    task_set = UserBehaviour
    min_wait = 5000
    max_wait = 60000

The above code wouldn't work when running Locust distributed, since the same code runs on each slave node, and they don't share any state. Therefore you would have to introduce some external datastore which the slave nodes could use to share states (e.g. PostgreSQL, redis, memcached or something else).

Answer 2

Alternatively, you can create users.py module to hold the users' information you need in your test cases, in my example, it holds email and cookies. Then you can call them randomly in your tasks. See below:

# locustfile.py
from locust import HttpLocust, TaskSet, task
from user_agent import *
from users import users_info


class UserBehaviour(TaskSet):
    def get_user(self):
        user = random.choice(users_info)
        return user

    @task(10)
    def get_siparislerim(self):
        user = self.get_user()
        user_agent = self.get_user_agent()
        r = self.client.get("/orders", headers = {"Cookie": user[1], 'User-Agent': user_agent})

class User(HttpLocust):
    task_set = UserBehaviour
    min_wait = 5000
    max_wait = 60000

User and user-agent can be called by a function. With this way, we could distribute the test with many users and different user-agents.

# users.py

users_info = [
['performancetest.1441926507@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926506@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926501@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926499@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926494@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926493@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926492@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926491@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926490@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926489@gmail.com', 'cookies_created_by_each_user'], 
['performancetest.1441926487@gmail.com', 'cookies_created_by_each_user']] 

Answer 3

Piggy-backing on @heyman's answer here. The example code will work, but continuing to start / stop tests will eventually clear out the USER_CREDENTIALS list, and start to throw errors.

I ended up adding the following:

2023 Update

From v1.3.0 the events API has changed.

USER_CREDENTIALS = generate_users()


@events.spawning_complete.add_listener
def spawn_complete_handler(**kw):
    global USER_CREDENTIALS
    USER_CREDENTIALS = generate_users()

Original answer

from locust import events # in addition to the other locust modules needed

def hatch_complete_handler(**kw):
    global USER_CREDENTIALS
    USER_CREDENTIALS = generate_users() # some function here to regenerate your list

events.hatch_complete += hatch_complete_handler

This refreshes your user list once your swarm finishes hatching.

Also keep in mind that you'll need a list longer than the number of users you wish to spawn.

Answer 4

I took a slightly different approach when implementing this for a distributed system. I utilized a very simple flask server that I made a get call to during the on_start portion of the TaskSet.

from flask import Flask, jsonify
app = Flask(__name__)

count = 0    #Shared Variable

@app.route("/")
def counter():
    global count

    count = count+1
    tenant = count // 5 + 1
    user = count % 5 + 1

    return jsonify({'count':count,'tenant':"load_tenant_{}".format(str(tenant)),'admin':"admin",'user':"load_user_{}".format(str(user))})

if __name__ == "__main__":
    app.run()

In this way I now have an endpoint I can get at http://localhost:5000/ on whatever host I run this. I just need to make this endpoint accessible to the slave systems and I wont have to worry about duplicate users or some type of round robin effect caused by having a limited set of user_info.