Simple Python 3.8+ tool to write load tests.
To create a load test, you need to create a Python module with some functions decorated with the scenario decorator – those functions needs to be coroutines. When executed, the function receives a session object inherited from aiohttp.ClientSession.
Here’s a full example :
""" This Molotov script has 2 scenario """ from molotov import scenario _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
When molotov runs, it creates some workers (coroutines) that will run the functions indefinitely until the test is over. A test is over when the time in seconds provided by –duration option is reached or when the maximum number of runs defined by –max-runs is reached.
By default, molotov will run for 86400 seconds (one day), but you can stop it any time with Ctrl-C or by sending a TERM signal to the process.
Each worker randomly picks one scenario to execute, given their weights. Once it’s finished, it picks the next one, and so on. In our example, scenario_two is picked ~60% of the time.
Check out aiohttp’s documentation to understand how to work with a session object.
To run the load script, you can provide its module name or its path. In the example below, the script is executed with 10 processes and 200 workers for 60 seconds:
$ molotov molotov/tests/example.py -p 10 -w 200 -d 60
When you run molotov, it will start a console application that will display the activity of each worker on the left panel. When a worker fails, the error will be displayed on the right panel
The footer will display a count of all successful and failed scenario runs, along with the current number of active workers (across all processes) and processes.
Since version 2.7, Molotov supports gRPC load testing.
To run against a gRPC server, you can use the session_kind option from the scenario decorator.
When the scenario is picked, a gRPC client will be created and used to execute the scenario. Each worker gets one session that will be reused for every scenario run. Molotov uses the standard grpcio client.
Here’s a full example :
""" This Molotov script runs against a GRPC endpoint """ from molotov import scenario from molotov.tests._grpc import helloworld_pb2, helloworld_pb2_grpc @scenario(weight=40) async def grpc_scenario( session, session_factory="grpc", grpc_url="ipv4:///127.0.0.1:50051" ): stub = helloworld_pb2_grpc.GreeterStub(session) response = await stub.SayHello(helloworld_pb2.HelloRequest(name="Alice")) session.print("Call done") assert response.message == "Hello, Alice!", response.message
Check out the detailed documentation:
- Command-line options
- API & Fixtures
- Extending Molotov
- Run from GitHub
- Run in Docker
- Step-by-step tutorial