Molotov

2.6 · active · verified Thu Apr 16

Molotov is a simple, Python 3.7+ tool for writing load tests. It is built upon `asyncio` and `aiohttp.client`, enabling highly concurrent testing scenarios. The current version is 2.6, released in October 2022. While there isn't a fixed release cadence, updates address Python compatibility, dependency changes, and new features.

Common errors

```
TypeError: scenario() takes 1 positional argument but 2 were given
```
cause Scenarios are asynchronous coroutines and must accept a `session` object as their first argument.

fix Define your scenario functions as `async def my_scenario(session): ...`
```
ModuleNotFoundError: No module named 'molotov'
```
cause Molotov is not installed in the current Python environment.

fix Run `pip install molotov` to install the library.
```
molotov.errors.MolotovError: Expected a coroutine
```
cause A function decorated with `@scenario` (or other async-expecting decorators) was defined as a regular function (`def`) instead of an asynchronous one (`async def`).

fix Change the function definition from `def` to `async def`.
```
certifi.errors.NoCACertificatesError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: unable to get local issuer certificate
```
cause Python's SSL module cannot find the necessary root SSL certificates to verify HTTPS connections, common in some Python installations or virtual environments.

fix Install `certifi` and ensure your Python environment is configured to use it, or run `pip install certifi` and re-run your tests. In some cases, you might need to run a specific `Install Certificates.command` on macOS.

Warnings

breaking Molotov version 2.5 dropped support for Python 3.6. Version 2.0 dropped support for Python 3.5. Ensure you are running Python 3.7 or superior for Molotov 2.6.
Fix: Upgrade your Python environment to 3.7+ (e.g., Python 3.10 for optimal compatibility with Molotov 2.5+).
breaking In Molotov 2.6, the internal multiprocessing library switched from `multiprocessing_on_dill` to `multiprocess`. While typically backward compatible, custom extensions or complex setups interacting directly with multiprocessing internals might require adjustments.
Fix: Review any custom multiprocessing logic in your test scripts; ensure compatibility with the `multiprocess` library.
deprecated The `aiomeasures` dependency was removed in Molotov 2.4 as the project is no longer maintained. Molotov now uses `aiodogstatsd` for statsd integration.
Fix: If you were relying on `aiomeasures`, migrate your metrics collection to use `aiodogstatsd` or other supported alternatives.
gotcha Molotov does not officially support Windows systems, especially concerning multiprocessing features. While basic functionality might work, distributed testing is not reliable.
Fix: Run Molotov on Linux or macOS environments for reliable load testing, or within a Docker container.
breaking Molotov 2.2 removed the `--disable-dns-resolver` option, as DNS resolution is now handled by `aiohttp`.
Fix: Remove `--disable-dns-resolver` from your Molotov command-line arguments. Configure `aiohttp` directly if specific DNS behavior is required.

Install

pip install molotov Install stable version

Imports

scenario
```
from molotov import scenario
```
The primary decorator for defining load test scenarios.
setup
```
from molotov import setup
```
Decorator for worker-level setup.
global_setup
```
from molotov import global_setup
```
Decorator for test-wide setup, executed once before processes and workers.
get_context
```
from molotov import get_context
```
Used to access session-specific data within scenarios or fixtures.

Quickstart

This quickstart defines two asynchronous scenarios that make GET requests to a target API. Each scenario receives an `aiohttp.ClientSession` object. The `weight` parameter controls the probability of a scenario being picked. To run, save this as `loadtest.py` and execute from the terminal.

import os
from molotov import scenario

_API = os.environ.get('MOLOTOV_TARGET_API', 'http://localhost:8080')

@scenario(weight=40)
async def scenario_one(session):
    async with session.get(_API) as resp:
        res = await resp.json()
        assert res["result"] == "OK"
        assert resp.status == 200

@scenario(weight=60)
async def scenario_two(session):
    async with session.get(_API) as resp:
        assert resp.status == 200

view raw JSON →