or run

npx @tessl/cli init
Log in

Version

Tile

Overview

Evals

Files

Files

docs

contrib.mddebugging.mdevents.mdexceptions.mdindex.mdload-shapes.mdtasksets.mduser-classes.mdwait-time.md

load-shapes.mddocs/

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

```