Using Locust as a library
It is possible to start a load test from your own Python code, instead of running Locust using the locust
command.
Start by creating an Environment
instance:
from locust.env import Environment
env = Environment(user_classes=[MyTestUser])
The Environment
instance’s
create_local_runner
,
create_master_runner
can then be used to start a
Runner
instance, which can be used to start a load test:
env.create_local_runner()
env.runner.start(5000, spawn_rate=20)
env.runner.greenlet.join()
It is also possible to bypass the dispatch and distribution logic, and manually control the spawned users:
new_users = env.runner.spawn_users({MyUserClass.__name__: 2})
new_users[1].my_custom_token = "custom-token-2"
new_users[0].my_custom_token = "custom-token-1"
The above example only works on standalone/local runner mode and is an experimental feature. A more common/better approach would be to use init
or test_start
Event hooks to get/create a list of tokens and use on_start and on_stop methods to read from that list and set them on your individual User instances.
Note
While it is possible to create locust workers this way (using create_worker_runner
), that almost never makes sense. Every worker needs to be in a separate Python process and interacting directly with the worker runner might break things. Just launch workers using the regular locust --worker ...
command.
We could also use the Environment
instance’s
create_web_ui
method to start a Web UI that can be used
to view the stats, and to control the runner (e.g. start and stop load tests):
env.create_local_runner()
env.create_web_ui()
env.web_ui.greenlet.join()
Skipping monkey patching
Some packages such as boto3 may have incompatibility when using Locust as a library, where monkey patching is already applied. In this case monkey patching may be disabled by setting LOCUST_SKIP_MONKEY_PATCH=1
as env variable.
Full example
#!/usr/bin/env python3
from locust import HttpUser, events, task
from locust.env import Environment
from locust.log import setup_logging
from locust.stats import stats_history, stats_printer
import gevent
setup_logging("INFO")
class MyUser(HttpUser):
host = "https://docs.locust.io"
@task
def t(self):
self.client.get("/")
# setup Environment and Runner
env = Environment(user_classes=[MyUser], events=events)
runner = env.create_local_runner()
# start a WebUI instance
web_ui = env.create_web_ui("127.0.0.1", 8089)
# execute init event handlers (only really needed if you have registered any)
env.events.init.fire(environment=env, runner=runner, web_ui=web_ui)
# start a greenlet that periodically outputs the current stats
gevent.spawn(stats_printer(env.stats))
# start a greenlet that save current stats to history
gevent.spawn(stats_history, env.runner)
# start the test
runner.start(1, spawn_rate=10)
# in 30 seconds stop the runner
gevent.spawn_later(30, runner.quit)
# wait for the greenlets
runner.greenlet.join()
# stop the web server for good measures
web_ui.stop()