0
# Load Test Shaping
1
2
LoadTestShape enables custom load patterns and shapes for advanced testing scenarios including ramp-up patterns, step functions, and complex load profiles over time.
3
4
## Capabilities
5
6
### LoadTestShape Base Class
7
8
Abstract base class for defining custom load test shapes and patterns.
9
10
```python { .api }
11
from locust import LoadTestShape
12
13
class LoadTestShape:
14
"""
15
Base class for defining custom load test shapes.
16
17
Attributes:
18
runner: Associated test runner instance
19
abstract (bool): Mark as abstract class
20
use_common_options (bool): Use common CLI options
21
"""
22
23
def tick(self):
24
"""
25
Define load pattern at current time.
26
27
Called every second during test execution to determine
28
the target user count and spawn rate.
29
30
Returns:
31
tuple: (user_count, spawn_rate) or (user_count, spawn_rate, user_classes)
32
None: End the test
33
34
Example:
35
return (100, 10) # 100 users, spawn 10 per second
36
return (50, 5, [UserA, UserB]) # 50 users from specific classes
37
return None # Stop the test
38
"""
39
40
def reset_time(self):
41
"""Reset the shape timer to start from beginning."""
42
43
def get_run_time(self):
44
"""
45
Get current test runtime.
46
47
Returns:
48
float: Runtime in seconds since shape started
49
"""
50
51
def get_current_user_count(self):
52
"""
53
Get current number of spawned users.
54
55
Returns:
56
int: Current user count
57
"""
58
```
59
60
## Usage Examples
61
62
### Step Load Pattern
63
64
```python
65
from locust import LoadTestShape, HttpUser, task, between
66
67
class StepLoadShape(LoadTestShape):
68
"""
69
Step load pattern with increasing user counts at intervals.
70
"""
71
72
step_time = 30 # 30 seconds per step
73
step_load = 10 # Add 10 users per step
74
spawn_rate = 5 # Spawn 5 users per second
75
time_limit = 300 # 5 minutes total
76
77
def tick(self):
78
run_time = self.get_run_time()
79
80
if run_time > self.time_limit:
81
return None # Stop test
82
83
current_step = run_time // self.step_time
84
user_count = int(current_step * self.step_load)
85
86
return (user_count, self.spawn_rate)
87
88
class WebUser(HttpUser):
89
wait_time = between(1, 3)
90
91
@task
92
def index(self):
93
self.client.get("/")
94
```
95
96
### Ramp Up/Down Pattern
97
98
```python
99
class RampUpDownShape(LoadTestShape):
100
"""
101
Ramp up to peak load, maintain, then ramp down.
102
"""
103
104
ramp_up_time = 120 # 2 minutes ramp up
105
peak_time = 300 # 5 minutes at peak
106
ramp_down_time = 120 # 2 minutes ramp down
107
peak_users = 100 # Peak user count
108
spawn_rate = 10 # Users per second
109
110
def tick(self):
111
run_time = self.get_run_time()
112
total_time = self.ramp_up_time + self.peak_time + self.ramp_down_time
113
114
if run_time >= total_time:
115
return None
116
117
if run_time < self.ramp_up_time:
118
# Ramp up phase
119
progress = run_time / self.ramp_up_time
120
user_count = int(self.peak_users * progress)
121
elif run_time < self.ramp_up_time + self.peak_time:
122
# Peak phase
123
user_count = self.peak_users
124
else:
125
# Ramp down phase
126
ramp_down_start = self.ramp_up_time + self.peak_time
127
ramp_down_progress = (run_time - ramp_down_start) / self.ramp_down_time
128
user_count = int(self.peak_users * (1 - ramp_down_progress))
129
130
return (user_count, self.spawn_rate)
131
```
132
133
### Spike Testing Pattern
134
135
```python
136
class SpikeTestShape(LoadTestShape):
137
"""
138
Spike testing with sudden load increases.
139
"""
140
141
base_users = 20
142
spike_users = 200
143
spike_duration = 60 # 1 minute spikes
144
normal_duration = 120 # 2 minutes normal
145
num_spikes = 3
146
spawn_rate = 20
147
148
def tick(self):
149
run_time = self.get_run_time()
150
cycle_time = self.spike_duration + self.normal_duration
151
total_time = cycle_time * self.num_spikes
152
153
if run_time >= total_time:
154
return None
155
156
cycle_position = run_time % cycle_time
157
158
if cycle_position < self.spike_duration:
159
# Spike phase
160
user_count = self.spike_users
161
else:
162
# Normal phase
163
user_count = self.base_users
164
165
return (user_count, self.spawn_rate)
166
```
167
168
### Time-Based Pattern
169
170
```python
171
import math
172
173
class SineWaveShape(LoadTestShape):
174
"""
175
Sine wave load pattern for cyclical testing.
176
"""
177
178
min_users = 10
179
max_users = 100
180
period = 300 # 5 minute cycle
181
spawn_rate = 5
182
test_duration = 1800 # 30 minutes
183
184
def tick(self):
185
run_time = self.get_run_time()
186
187
if run_time > self.test_duration:
188
return None
189
190
# Calculate sine wave
191
amplitude = (self.max_users - self.min_users) / 2
192
mid_point = self.min_users + amplitude
193
wave_position = 2 * math.pi * run_time / self.period
194
user_count = int(mid_point + amplitude * math.sin(wave_position))
195
196
return (user_count, self.spawn_rate)
197
```
198
199
### User Class Distribution
200
201
```python
202
class MultiUserShape(LoadTestShape):
203
"""
204
Shape with different user class distributions over time.
205
"""
206
207
def tick(self):
208
run_time = self.get_run_time()
209
210
if run_time > 600: # 10 minutes
211
return None
212
213
if run_time < 180: # First 3 minutes: only regular users
214
return (50, 10, [RegularUser])
215
elif run_time < 360: # Next 3 minutes: mixed users
216
return (75, 10, [RegularUser, PowerUser])
217
else: # Last 4 minutes: all user types
218
return (100, 10, [RegularUser, PowerUser, AdminUser])
219
220
class RegularUser(HttpUser):
221
weight = 3
222
wait_time = between(2, 5)
223
224
@task
225
def browse(self):
226
self.client.get("/")
227
228
class PowerUser(HttpUser):
229
weight = 2
230
wait_time = between(1, 3)
231
232
@task
233
def advanced_features(self):
234
self.client.get("/admin/dashboard")
235
236
class AdminUser(HttpUser):
237
weight = 1
238
wait_time = between(3, 8)
239
240
@task
241
def admin_tasks(self):
242
self.client.get("/admin/users")
243
```
244
245
### Conditional Load Patterns
246
247
```python
248
class ConditionalShape(LoadTestShape):
249
"""
250
Load shape that adapts based on response times.
251
"""
252
253
target_response_time = 2000 # 2 seconds
254
max_users = 200
255
min_users = 10
256
adjustment_rate = 10
257
258
def tick(self):
259
run_time = self.get_run_time()
260
261
if run_time > 1800: # 30 minutes max
262
return None
263
264
current_users = self.get_current_user_count()
265
266
# Get average response time from stats
267
stats = self.runner.environment.stats
268
if stats.total.num_requests > 0:
269
avg_response_time = stats.total.avg_response_time
270
271
if avg_response_time > self.target_response_time:
272
# Response time too high, reduce users
273
new_users = max(self.min_users, current_users - self.adjustment_rate)
274
elif avg_response_time < self.target_response_time * 0.5:
275
# Response time good, increase users
276
new_users = min(self.max_users, current_users + self.adjustment_rate)
277
else:
278
# Response time acceptable, maintain
279
new_users = current_users
280
else:
281
# No stats yet, start with minimum
282
new_users = self.min_users
283
284
return (new_users, 5)
285
```
286
287
### Schedule-Based Pattern
288
289
```python
290
import datetime
291
292
class ScheduleBasedShape(LoadTestShape):
293
"""
294
Load pattern based on business hours schedule.
295
"""
296
297
def get_business_hours_load(self, hour):
298
"""Get user count based on business hour."""
299
if 9 <= hour <= 17: # Business hours
300
if 12 <= hour <= 13: # Lunch hour peak
301
return 150
302
elif hour in [9, 17]: # Start/end of day
303
return 100
304
else: # Regular business hours
305
return 120
306
elif 18 <= hour <= 22: # Evening
307
return 60
308
else: # Night/early morning
309
return 20
310
311
def tick(self):
312
run_time = self.get_run_time()
313
314
if run_time > 86400: # 24 hours max
315
return None
316
317
# Simulate time-based load
318
current_hour = int((run_time / 3600) % 24)
319
user_count = self.get_business_hours_load(current_hour)
320
321
return (user_count, 10)
322
```
323
324
## Types
325
326
```python { .api }
327
from typing import Optional, Tuple, List, Type, Union
328
from locust import User
329
330
# Shape return types
331
UserCount = int
332
SpawnRate = int
333
UserClasses = List[Type[User]]
334
335
# Shape tick return types
336
ShapeReturn = Union[
337
Tuple[UserCount, SpawnRate],
338
Tuple[UserCount, SpawnRate, UserClasses],
339
None # Stop test
340
]
341
342
# Load shape configuration
343
class LoadTestShape:
344
def tick(self) -> Optional[ShapeReturn]: ...
345
def reset_time(self) -> None: ...
346
def get_run_time(self) -> float: ...
347
def get_current_user_count(self) -> int: ...
348
```