0
# Debugging and Utilities
1
2
Locust provides debugging tools and utilities for development-time testing, task analysis, and troubleshooting load test scenarios.
3
4
## Capabilities
5
6
### Single User Debugging
7
8
Function for running a single user instance to debug test scenarios without full load test execution.
9
10
```python { .api }
11
from locust.debug import run_single_user
12
13
def run_single_user(user_class, include_length=False, include_time=False,
14
include_context=False, include_payload=False, loglevel="WARNING"):
15
"""
16
Run a single user instance for debugging purposes.
17
18
Executes one user instance to test task behavior, request patterns,
19
and response handling without starting a full load test.
20
21
Args:
22
user_class: User class to run
23
include_length (bool): Include response length in output
24
include_time (bool): Include timing information in output
25
include_context (bool): Include request context information
26
include_payload (bool): Include request/response payload data
27
loglevel (str): Logging level ("DEBUG", "INFO", "WARNING", "ERROR")
28
29
Usage:
30
run_single_user(MyHttpUser, include_time=True, loglevel="INFO")
31
"""
32
```
33
34
### Task Ratio Analysis
35
36
Functions for analyzing and inspecting task execution ratios and distributions.
37
38
```python { .api }
39
from locust.user.inspectuser import print_task_ratio, print_task_ratio_json, get_ratio
40
41
def print_task_ratio(user_classes, num_users, total):
42
"""
43
Print task execution ratios for user classes.
44
45
Displays the relative frequency and probability of task execution
46
for analysis and debugging of task distribution.
47
48
Args:
49
user_classes (list): List of User classes to analyze
50
num_users (int): Total number of users to distribute
51
total (bool): Whether to show total ratios across all users
52
"""
53
54
def print_task_ratio_json(user_classes, num_users):
55
"""
56
Print task execution ratios in JSON format.
57
58
Outputs task ratio analysis as structured JSON data
59
for programmatic processing and integration.
60
61
Args:
62
user_classes (list): List of User classes to analyze
63
num_users (int): Total number of users to distribute
64
"""
65
66
def get_ratio(user_classes, user_spawned, total):
67
"""
68
Get task execution ratios as data structure.
69
70
Returns task ratio analysis as dictionary for
71
programmatic inspection and processing.
72
73
Args:
74
user_classes (list): List of User classes to analyze
75
user_spawned (dict): Dict mapping user class names to spawned counts
76
total (bool): Whether to calculate total ratios across all users
77
78
Returns:
79
dict: Task ratio analysis data with nested task ratios
80
"""
81
```
82
83
## Usage Examples
84
85
### Basic Single User Debugging
86
87
```python
88
from locust import HttpUser, task, between
89
from locust.debug import run_single_user
90
91
class DebugUser(HttpUser):
92
wait_time = between(1, 3)
93
host = "http://localhost:8000"
94
95
def on_start(self):
96
print("User starting - performing login")
97
response = self.client.post("/login", json={
98
"username": "testuser",
99
"password": "secret"
100
})
101
print(f"Login response: {response.status_code}")
102
103
@task(3)
104
def browse_pages(self):
105
print("Browsing pages task")
106
pages = ["/", "/about", "/products"]
107
for page in pages:
108
response = self.client.get(page)
109
print(f"Page {page}: {response.status_code}")
110
111
@task(1)
112
def api_call(self):
113
print("API call task")
114
response = self.client.get("/api/data")
115
print(f"API response: {response.status_code}, Length: {len(response.content)}")
116
117
def on_stop(self):
118
print("User stopping - performing logout")
119
self.client.post("/logout")
120
121
if __name__ == "__main__":
122
# Debug the user behavior
123
print("=== Running Single User Debug ===")
124
run_single_user(DebugUser, include_time=True, include_length=True, loglevel="INFO")
125
```
126
127
### Advanced Single User Debugging
128
129
```python
130
from locust import HttpUser, TaskSet, task, between
131
from locust.debug import run_single_user
132
import json
133
134
class APITestSet(TaskSet):
135
def on_start(self):
136
print("Starting API test set")
137
# Authenticate
138
response = self.client.post("/auth", json={
139
"client_id": "test_client",
140
"client_secret": "test_secret"
141
})
142
self.token = response.json().get("access_token")
143
print(f"Got auth token: {self.token[:20]}...")
144
145
@task
146
def test_endpoint_a(self):
147
print("Testing endpoint A")
148
headers = {"Authorization": f"Bearer {self.token}"}
149
response = self.client.get("/api/endpoint-a", headers=headers)
150
print(f"Endpoint A: {response.status_code}")
151
152
# Debug response content
153
if response.status_code != 200:
154
print(f"Error response: {response.text}")
155
156
@task
157
def test_endpoint_b(self):
158
print("Testing endpoint B with payload")
159
headers = {"Authorization": f"Bearer {self.token}"}
160
payload = {"test_data": "debug_value", "timestamp": time.time()}
161
162
response = self.client.post("/api/endpoint-b",
163
json=payload,
164
headers=headers)
165
print(f"Endpoint B: {response.status_code}")
166
167
# Debug response validation
168
try:
169
result = response.json()
170
if "status" in result:
171
print(f"API Status: {result['status']}")
172
except json.JSONDecodeError:
173
print("Response is not valid JSON")
174
175
class DebugAPIUser(HttpUser):
176
wait_time = between(1, 2)
177
host = "http://api.localhost:8000"
178
tasks = [APITestSet]
179
180
if __name__ == "__main__":
181
# Comprehensive debugging with all options
182
print("=== Comprehensive API Debug ===")
183
run_single_user(
184
DebugAPIUser,
185
include_length=True,
186
include_time=True,
187
include_context=True,
188
include_payload=True,
189
loglevel="DEBUG"
190
)
191
```
192
193
### Task Ratio Analysis
194
195
```python
196
from locust import HttpUser, TaskSet, task, between
197
from locust.user.inspectuser import print_task_ratio, print_task_ratio_json, get_ratio
198
199
class ShoppingTaskSet(TaskSet):
200
@task(5) # 5x weight
201
def browse_products(self):
202
self.client.get("/products")
203
204
@task(3) # 3x weight
205
def view_product(self):
206
self.client.get("/product/123")
207
208
@task(1) # 1x weight
209
def add_to_cart(self):
210
self.client.post("/cart/add")
211
212
class UserBehaviorA(HttpUser):
213
wait_time = between(1, 3)
214
tasks = [ShoppingTaskSet]
215
weight = 3 # 3x more likely to be selected
216
217
class UserBehaviorB(HttpUser):
218
wait_time = between(2, 5)
219
weight = 1 # 1x likelihood
220
221
@task(2)
222
def admin_panel(self):
223
self.client.get("/admin")
224
225
@task(1)
226
def reports(self):
227
self.client.get("/reports")
228
229
if __name__ == "__main__":
230
user_classes = [UserBehaviorA, UserBehaviorB]
231
num_users = 100 # Total users to analyze
232
233
print("=== Task Ratio Analysis ===")
234
print_task_ratio(user_classes, num_users, total=True)
235
236
print("\n=== Task Ratio JSON ===")
237
print_task_ratio_json(user_classes, num_users)
238
239
print("\n=== Programmatic Analysis ===")
240
# Calculate user distribution based on weights
241
user_spawned = {"UserBehaviorA": 75, "UserBehaviorB": 25} # Example distribution
242
ratios = get_ratio(user_classes, user_spawned, total=True)
243
244
# Process ratio data
245
for user_class, data in ratios.items():
246
print(f"\nUser Class: {user_class}")
247
print(f"Selection Weight: {data.get('weight', 1)}")
248
print("Tasks:")
249
for task_name, task_ratio in data.get('tasks', {}).items():
250
print(f" {task_name}: {task_ratio:.2%}")
251
```
252
253
### Development-Time Testing
254
255
```python
256
from locust import HttpUser, task, between, events
257
from locust.debug import run_single_user
258
import time
259
260
class DevelopmentUser(HttpUser):
261
wait_time = between(0.5, 1) # Fast for development
262
host = "http://localhost:3000"
263
264
def __init__(self, *args, **kwargs):
265
super().__init__(*args, **kwargs)
266
self.start_time = time.time()
267
self.request_count = 0
268
269
def on_start(self):
270
print("π Starting development test")
271
print(f"Target host: {self.host}")
272
273
# Check if server is responding
274
try:
275
response = self.client.get("/health")
276
if response.status_code == 200:
277
print("β Server health check passed")
278
else:
279
print(f"β οΈ Server health check failed: {response.status_code}")
280
except Exception as e:
281
print(f"β Server health check error: {e}")
282
283
@task(3)
284
def test_main_page(self):
285
"""Test main application page"""
286
print("Testing main page...")
287
start = time.time()
288
289
response = self.client.get("/")
290
duration = (time.time() - start) * 1000
291
292
self.request_count += 1
293
print(f"Main page: {response.status_code} ({duration:.1f}ms)")
294
295
# Development-specific checks
296
if duration > 2000:
297
print("β οΈ Slow response detected!")
298
299
if response.status_code != 200:
300
print(f"β Unexpected status: {response.status_code}")
301
print(f"Response: {response.text[:200]}...")
302
303
@task(2)
304
def test_api_endpoint(self):
305
"""Test API endpoint functionality"""
306
print("Testing API endpoint...")
307
308
# Test with different parameters
309
test_params = [
310
{"id": 1, "type": "user"},
311
{"id": 2, "type": "admin"},
312
{"id": 999, "type": "invalid"} # Test edge case
313
]
314
315
for params in test_params:
316
response = self.client.get("/api/data", params=params)
317
print(f"API test {params}: {response.status_code}")
318
319
# Validate response structure in development
320
if response.status_code == 200:
321
try:
322
data = response.json()
323
required_fields = ["id", "status", "data"]
324
missing_fields = [f for f in required_fields if f not in data]
325
if missing_fields:
326
print(f"β οΈ Missing fields: {missing_fields}")
327
except json.JSONDecodeError:
328
print("β Invalid JSON response")
329
330
@task(1)
331
def test_form_submission(self):
332
"""Test form submission"""
333
print("Testing form submission...")
334
335
test_data = {
336
"name": "Test User",
337
"email": "test@example.com",
338
"message": "Development test message"
339
}
340
341
response = self.client.post("/contact", json=test_data)
342
print(f"Form submission: {response.status_code}")
343
344
# Check for validation errors in development
345
if response.status_code == 400:
346
try:
347
errors = response.json().get("errors", [])
348
print(f"Validation errors: {errors}")
349
except:
350
print("Could not parse validation errors")
351
352
def on_stop(self):
353
duration = time.time() - self.start_time
354
print(f"\nπ Development Test Summary:")
355
print(f"Duration: {duration:.1f} seconds")
356
print(f"Requests: {self.request_count}")
357
print(f"Rate: {self.request_count/duration:.1f} req/sec")
358
print("π Development test completed")
359
360
if __name__ == "__main__":
361
print("=== Development Testing Mode ===")
362
run_single_user(
363
DevelopmentUser,
364
include_time=True,
365
include_length=True,
366
loglevel="INFO"
367
)
368
```
369
370
### Custom Debugging Tools
371
372
```python
373
from locust import HttpUser, task, between
374
from locust.debug import run_single_user
375
import json
376
import time
377
378
class DebuggingMixin:
379
"""Mixin for adding debugging capabilities to users"""
380
381
def debug_request(self, method, url, **kwargs):
382
"""Enhanced request method with debugging"""
383
print(f"π {method.upper()} {url}")
384
385
# Log request details
386
if 'json' in kwargs:
387
print(f" Request JSON: {json.dumps(kwargs['json'], indent=2)}")
388
if 'data' in kwargs:
389
print(f" Request Data: {kwargs['data']}")
390
if 'headers' in kwargs:
391
print(f" Headers: {kwargs['headers']}")
392
393
start_time = time.time()
394
response = getattr(self.client, method.lower())(url, **kwargs)
395
duration = (time.time() - start_time) * 1000
396
397
# Log response details
398
print(f" β±οΈ {duration:.1f}ms | π {response.status_code} | π {len(response.content)} bytes")
399
400
if response.status_code >= 400:
401
print(f" β Error Response: {response.text[:200]}...")
402
403
# Try to parse JSON response
404
try:
405
json_response = response.json()
406
print(f" π Response JSON: {json.dumps(json_response, indent=2)[:300]}...")
407
except:
408
print(f" π Response Text: {response.text[:100]}...")
409
410
return response
411
412
def debug_wait(self, message="Waiting"):
413
"""Debug-friendly wait with message"""
414
wait_time = self.wait_time() if callable(self.wait_time) else self.wait_time
415
print(f"β³ {message} for {wait_time:.1f} seconds...")
416
time.sleep(wait_time)
417
418
class DebugHTTPUser(DebuggingMixin, HttpUser):
419
"""HTTP User with debugging capabilities"""
420
421
wait_time = between(1, 2)
422
host = "http://localhost:8000"
423
424
@task
425
def debug_workflow(self):
426
"""Example workflow with debugging"""
427
print("\nπ― Starting debug workflow")
428
429
# Step 1: Login
430
print("Step 1: Login")
431
login_response = self.debug_request("post", "/login", json={
432
"username": "debug_user",
433
"password": "debug_pass"
434
})
435
436
if login_response.status_code == 200:
437
token = login_response.json().get("token")
438
headers = {"Authorization": f"Bearer {token}"}
439
440
# Step 2: Get user data
441
print("Step 2: Get user data")
442
self.debug_request("get", "/api/user/me", headers=headers)
443
444
# Step 3: Update profile
445
print("Step 3: Update profile")
446
self.debug_request("put", "/api/user/profile",
447
json={"name": "Debug User Updated"},
448
headers=headers)
449
450
print("β Debug workflow completed\n")
451
self.debug_wait("Workflow complete, waiting")
452
453
# Interactive debugging session
454
def interactive_debug():
455
"""Interactive debugging session"""
456
print("=== Interactive Debugging Session ===")
457
458
while True:
459
print("\nOptions:")
460
print("1. Run single user")
461
print("2. Analyze task ratios")
462
print("3. Run with detailed logging")
463
print("4. Exit")
464
465
choice = input("Select option (1-4): ").strip()
466
467
if choice == "1":
468
print("Running single user...")
469
run_single_user(DebugHTTPUser, include_time=True, loglevel="INFO")
470
471
elif choice == "2":
472
print("Analyzing task ratios...")
473
print_task_ratio([DebugHTTPUser], num_users=10, total=True)
474
475
elif choice == "3":
476
print("Running with detailed logging...")
477
run_single_user(DebugHTTPUser,
478
include_time=True,
479
include_length=True,
480
include_context=True,
481
include_payload=True,
482
loglevel="DEBUG")
483
484
elif choice == "4":
485
print("Exiting debug session")
486
break
487
488
else:
489
print("Invalid option")
490
491
if __name__ == "__main__":
492
interactive_debug()
493
```
494
495
## Types
496
497
```python { .api }
498
from typing import Dict, List, Any, Type, Optional
499
from locust import User
500
501
# Debugging function types
502
UserClass = Type[User]
503
UserClassList = List[UserClass]
504
LogLevel = str # "DEBUG", "INFO", "WARNING", "ERROR"
505
506
# Task ratio analysis types
507
TaskRatioData = Dict[str, Any]
508
TaskRationResult = Dict[str, TaskRatioData]
509
510
# Debug function signatures
511
def run_single_user(
512
user_class: UserClass,
513
include_length: bool = False,
514
include_time: bool = False,
515
include_context: bool = False,
516
include_payload: bool = False,
517
loglevel: LogLevel = "WARNING"
518
) -> None: ...
519
520
def print_task_ratio(user_classes: UserClassList) -> None: ...
521
def print_task_ratio_json(user_classes: UserClassList) -> None: ...
522
def get_ratio(user_classes: UserClassList) -> TaskRationResult: ...
523
```