Testing other systems using custom clients

Locust was built with HTTP as its main target. However, it can easily be extended to load test any request/response based system, by writing a custom client that triggers request_success and request_failure events.


Any protocol libraries that you use must be gevent-friendly (use the Python socket module or some other standard library function like subprocess), or your calls will block the whole Locust process.

Some C libraries cannot be monkey patched by gevent, but allow for other workarounds. For example, if you want to use psycopg2 to performance test PostgreSQL, can use psycogreen.

Sample XML-RPC User client

Here is an example of a User class, XmlRpcUser, which provides an XML-RPC client, XmlRpcUser, and tracks all requests made:

import time
from xmlrpc.client import ServerProxy, Fault

from locust import User, task, between

class XmlRpcClient(ServerProxy):
    Simple, sample XML RPC client implementation that wraps xmlrpclib.ServerProxy and
    fires locust events on request_success and request_failure, so that all requests
    gets tracked in locust's statistics.

    _locust_environment = None

    def __getattr__(self, name):
        func = ServerProxy.__getattr__(self, name)

        def wrapper(*args, **kwargs):
            start_time = time.time()
                result = func(*args, **kwargs)
            except Fault as e:
                total_time = int((time.time() - start_time) * 1000)
                    request_type="xmlrpc", name=name, response_time=total_time, exception=e
                total_time = int((time.time() - start_time) * 1000)
                    request_type="xmlrpc", name=name, response_time=total_time, response_length=0
                # In this example, I've hardcoded response_length=0. If we would want the response length to be
                # reported correctly in the statistics, we would probably need to hook in at a lower level

        return wrapper

class XmlRpcUser(User):
    This is the abstract User class which should be subclassed. It provides an XML-RPC client
    that can be used to make XML-RPC requests that will be tracked in Locust's statistics.

    abstract = True

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.client = XmlRpcClient(self.host)
        self.client._locust_environment = self.environment

class ApiUser(XmlRpcUser):
    host = ""
    wait_time = between(0.1, 1)

    def get_time(self):

    def get_random_number(self):
        self.client.get_random_number(0, 100)

If you’ve written Locust tests before, you’ll recognize the class called ApiUser which is a normal User class that has a couple of tasks declared. However, the ApiUser inherits from XmlRpcUser that you can see right above ApiUser. The XmlRpcUser is marked as abstract using abstract = True which means that Locust will not try to create simulated users from that class (only of classes that extend it). XmlRpcUser provides an instance of XmlRpcClient under the client attribute.

The XmlRpcClient is a wrapper around the standard library’s xmlrpc.client.ServerProxy. It basically just proxies the function calls, but with the important addition of firing locust.event.Events.request_success and locust.event.Events.request_failure events, which will record all calls in Locust’s statistics.

Here’s an implementation of an XML-RPC server that would work as a server for the code above:

import random
import time
from xmlrpc.server import SimpleXMLRPCServer

def get_time():
    return time.time()

def get_random_number(low, high):
    return random.randint(low, high)

server = SimpleXMLRPCServer(("localhost", 8877))
print("Listening on port 8877...")
server.register_function(get_time, "get_time")
server.register_function(get_random_number, "get_random_number")

For more examples, see locust-plugins