Locust Directory

Testing websockets

How to use LoadForge to load test websockets based applications


Websocket testing is naturally more complicated than simple webserver, or web application load testing. However, with LoadForge and locust files you can connect to and load test websockets easily.


import time
import json
import gevent

from uuid import uuid4

from locust import HttpLocust, TaskSet, task, events, User
import websocket

class SocketClient(object):
    def __init__(self, host): = host
        self.session_id = uuid4().hex

    def connect(self): = websocket.WebSocket()

        events.quitting += self.on_close


    def attach_session(self):
        payload = {'action': 'attach_session', 'session_id': self.session_id}
        json_data = self.send_with_response(payload)

        assert json_data['action'].lower() == 'attach_session'
        assert json_data['message_status'].lower() == 'ok'
        assert json_data['session_id'] == self.session_id

    def send_with_response(self, payload):
        json_data = json.dumps(payload)

        g = gevent.spawn(, json_data)
        g.get(block=True, timeout=2)
        g = gevent.spawn(
        result = g.get(block=True, timeout=10)

        json_data = json.loads(result)
        return json_data

    def on_close(self):

    def send(self, payload):
        message_id = uuid4().hex
        payload.update({'_messageId': message_id,
                        'session_id': self.session_id})
        start_time = time.time()
        e = None
            data = self.send_with_response(payload)
            assert data['_messageId'] == message_id
            assert data['session_id'] == self.session_id
        except AssertionError as exp:
            e = exp
        except Exception as exp:
            e = exp
        elapsed = int((time.time() - start_time) * 1000)
        if e:
  'sockjs', name='send',
                                        response_time=elapsed, exception=e)
  'sockjs', name='send',

class WSBehavior(TaskSet):
    def action(self):
        data = {
            "action": "do_stuff",
            "param": "123",

class WSUser(User):
    task_set = WSBehavior
    min_wait = 1000
    max_wait = 3000

    def __init__(self, *args, **kwargs):
        super(WSUser, self).__init__(*args, **kwargs)
        self.client = SocketClient('ws://%s/rt/websocket' %
This guide is part of the LoadForge Directory, an index of locustfile's for use with LoadForge website and API load tests. We also provide a wizard to generate tests, and onboarding assistance for clients. Contact us should you have any questions.

LoadForge provides load testing and stress tests for websites, APIs and servers. It uses your cloud account to rapidly scale large numbers of simulated users to load test your website, store, API, or application for cheap - just cents per test!

Ready to run your test?
Start your first test within minutes.