Installation
- pip install locustInstall via pip
- pip install locust[all]With extras
- locust --versionCheck version
- locust --helpShow help
Basic Script Structure
from locust import HttpUser, task, between
class WebsiteUser(HttpUser):
wait_time = between(1, 3) # Wait 1-3 seconds between tasks
@task
def index(self):
self.client.get("/")
@task(3) # 3x more likely to be picked
def about(self):
self.client.get("/about")
CLI Commands
- locustStart web UI (8089)
- locust -f file.pySpecify file
- --headlessNo web UI
- -u 100Number of users
- -r 10Spawn rate/sec
- -t 60sRun time
- -H http://hostTarget host
HTTP Methods
- self.client.get()GET request
- self.client.post()POST request
- self.client.put()PUT request
- self.client.delete()DELETE request
- self.client.patch()PATCH request
- self.client.head()HEAD request
Wait Time Functions
- between(1, 5)Random 1-5 sec
- constant(2)Fixed 2 sec
- constant_pacing(5)5 sec total cycle
- constant_throughput(0.5)0.5 tasks/sec
Lifecycle Events
- on_start()User starts
- on_stop()User stops
- @events.initLocust init
- @events.test_startTest begins
- @events.test_stopTest ends
- @events.requestEach request
POST with JSON Data
@task
def create_user(self):
payload = {"username": "test", "email": "test@example.com"}
headers = {"Content-Type": "application/json"}
with self.client.post("/api/users", json=payload, headers=headers,
catch_response=True) as response:
if response.status_code == 201:
response.success()
else:
response.failure(f"Failed: {response.status_code}")
Distributed Mode
# Start master node
locust -f locustfile.py --master
# Start worker nodes (run on multiple machines)
locust -f locustfile.py --worker --master-host=192.168.1.100
# Or with expected workers
locust -f locustfile.py --master --expect-workers=4
Response Validation
with self.client.get("/api",
catch_response=True) as r:
if "success" in r.text:
r.success()
else:
r.failure("Missing data")
TaskSet Classes
from locust import TaskSet
class UserBehavior(TaskSet):
@task
def browse(self):
self.client.get("/browse")
@task
def stop(self):
self.interrupt()