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
or
create_worker_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 mode and is an experimental feature, meaning that it could be removed in future versions. But it’s useful if you want to have fine-grained control on the spawned users.
Note
Do not attempt to create a master runner and worker(s) in the same Python process. It does not work, and even if it did, it would not give you better performance than running a single LocalRunner. Every worker must run in its own process, there is no way around that.
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()
Full example¶
import gevent
from locust import HttpUser, task, between
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging
setup_logging("INFO", None)
class User(HttpUser):
wait_time = between(1, 3)
host = "https://docs.locust.io"
@task
def my_task(self):
self.client.get("/")
@task
def task_404(self):
self.client.get("/non-existing-path")
# setup Environment and Runner
env = Environment(user_classes=[User])
env.create_local_runner()
# start a WebUI instance
env.create_web_ui("127.0.0.1", 8089)
# 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
env.runner.start(1, spawn_rate=10)
# in 60 seconds stop the runner
gevent.spawn_later(60, lambda: env.runner.quit())
# wait for the greenlets
env.runner.greenlet.join()
# stop the web server for good measures
env.web_ui.stop()