Locust Plugins

5.0.0 · active · verified Sat Apr 11

Locust Plugins is a collection of useful extensions for Locust, the open-source load testing tool. It provides implementations for various non-HTTP protocols like Playwright, Selenium/Webdriver, Kafka, FTP, and Telnet 3270, as well as enhanced reporting capabilities for TimescaleDB, Grafana, and Azure Application Insights. The library is currently at version 5.0.0 and maintains an active development cycle, often aligning with updates to Locust core.

Warnings

breaking As of `locust-plugins` version 5.0.0, the `MqttUser`, `SocketIOUser`, and `RestUser` classes have been removed from this library. They are now part of Locust core.
Fix: Update your imports to `from locust.contrib.mqtt import MqttUser`, `from locust.contrib.socketio import SocketIOUser`, and `from locust.contrib.rest import RestUser` respectively. Ensure your Locust version is compatible (Locust 2.41.0+ for MqttUser, 2.39.0+ for SocketIOUser).
breaking The `--profile` command-line argument was removed in `locust-plugins` 4.7.0. It has been integrated into Locust core functionality.
Fix: Refer to the official Locust documentation for the updated `--profile` argument usage directly within Locust.
gotcha User types like `PlaywrightUser` and `WebdriverUser` (Selenium) are resource-intensive. Running too many concurrent browser instances per worker can lead to high CPU usage and degraded performance.
Fix: Optimize tests by blocking unnecessary resources (e.g., images, third-party scripts). Scale by using more Locust worker instances across multiple machines instead of increasing the number of users per worker (e.g., limit to 4-5 browser users per worker).
gotcha When implementing custom non-HTTP User classes (or using older versions of `SocketIOUser`), avoid directly calling blocking receive methods (e.g., `self.ws.recv()`).
Fix: Instead, implement event-driven callbacks (e.g., `on_message()`) to handle incoming data. This aligns with Locust's gevent-based concurrency model and prevents blocking the entire worker process.
gotcha For logging and graphing results using TimescaleDB and Grafana, ensure you configure the `--timescale` argument when running Locust and have a TimescaleDB instance set up.
Fix: Follow the setup instructions in the `locust-plugins` documentation for Timescale/Grafana integration, typically involving `docker-compose` for a quick setup. Activate with `locust --timescale ...`

Install

pip install locust-plugins Install stable version
pip install 'locust-plugins[all]' Install with all optional dependencies (e.g., Playwright, Selenium)

Imports

PlaywrightUser

from locust_plugins.users.playwright import PlaywrightUser

Timescale

from locust_plugins.listeners.timescale import Timescale

ApplicationInsights

from locust_plugins.listeners.appinsights import ApplicationInsights

FtpUser

from locust_plugins.users.ftpuser import FtpUser

MqttUser
```
from locust.contrib.mqtt import MqttUser
```
MqttUser was moved into Locust core in locust-plugins 5.0.0 and Locust 2.41.0. Import directly from `locust.contrib.mqtt` instead.
SocketIOUser
```
from locust.contrib.socketio import SocketIOUser
```
SocketIOUser was moved into Locust core in locust-plugins 5.0.0 and Locust 2.39.0. Import directly from `locust.contrib.socketio` instead.
RestUser
```
from locust.contrib.rest import RestUser
```
RestUser was moved into Locust core in locust-plugins 5.0.0. Import directly from `locust.contrib.rest` instead.

Quickstart

This quickstart demonstrates a basic PlaywrightUser, launching a browser, navigating to Google, performing a search, and taking a screenshot. It then manually fires a request event for Locust statistics.

from locust import task, between
from locust_plugins.users.playwright import PlaywrightUser

class MyPlaywrightUser(PlaywrightUser):
    host = "https://www.google.com"
    wait_time = between(2, 5)

    @task
    async def search_for_locust(self):
        page = self.page  # Access the Playwright page object
        await page.goto(self.host)
        await page.fill('textarea[name="q"]', "Locust load testing")
        await page.press('textarea[name="q"]', 'Enter')
        await page.wait_for_selector("#search"); # Wait for search results
        await page.screenshot(path="locust_search.png")
        self.environment.events.request.fire(
            request_type="Playwright",
            name="/search",
            response_time=1000, # Dummy response time
            response_length=100,
            context={
                "user": self.id
            },
            exception=None,
        )

view raw JSON →