Skip to main content

​
Overview

GraphQL subscriptions provide real-time data streams over WebSockets. LoadForge supports subscription load testing by leveraging Locust and the WebSocketUser from locust-plugins.

​
Locust Test Script (locust.py)

# locust.py
import json
from locust import between, task
from locust_plugins.users import WebSocketUser

class GraphqlSubsUser(WebSocketUser):
    wait_time = between(1, 3)
    host = "wss://api.yourdomain.com/graphql"

    def on_start(self):
        # Open WebSocket connection
        self.connect("/graphql")

        # Initialize the subscription
        init_msg = {"type": "connection_init", "payload": {}}
        self.send(json.dumps(init_msg))
        self.receive()

        # Start subscription
        sub_msg = {
            "id": "1",
            "type": "start",
            "payload": {"query": "subscription { newMessage { content sender } }"}
        }
        self.send(json.dumps(sub_msg))

    @task
    def listen(self):
        # Receive a subscription event
        event = self.receive()
        data = json.loads(event)
        print(data)
Notes:
  • Install dependencies: pip install locust locust-plugins websockets if you wish to test locally.
  • Replace host and GraphQL query with your subscription endpoint.
⌘I