tessl/pypi-locust
Developer-friendly load testing framework for HTTP and other protocols with distributed testing capabilities.
—
Pending
load-shapes.mddocs/
Load Test Shaping
LoadTestShape enables custom load patterns and shapes for advanced testing scenarios including ramp-up patterns, step functions, and complex load profiles over time.
Capabilities
LoadTestShape Base Class
Abstract base class for defining custom load test shapes and patterns.
from locust import LoadTestShape
class LoadTestShape:
"""
Base class for defining custom load test shapes.
Attributes:
runner: Associated test runner instance
abstract (bool): Mark as abstract class
use_common_options (bool): Use common CLI options
"""
def tick(self):
"""
Define load pattern at current time.
Called every second during test execution to determine
the target user count and spawn rate.
Returns:
tuple: (user_count, spawn_rate) or (user_count, spawn_rate, user_classes)
None: End the test
Example:
return (100, 10) # 100 users, spawn 10 per second
return (50, 5, [UserA, UserB]) # 50 users from specific classes
return None # Stop the test
"""
def reset_time(self):
"""Reset the shape timer to start from beginning."""
def get_run_time(self):
"""
Get current test runtime.
Returns:
float: Runtime in seconds since shape started
"""
def get_current_user_count(self):
"""
Get current number of spawned users.
Returns:
int: Current user count
"""Usage Examples
Step Load Pattern
from locust import LoadTestShape, HttpUser, task, between
class StepLoadShape(LoadTestShape):
"""
Step load pattern with increasing user counts at intervals.
"""
step_time = 30 # 30 seconds per step
step_load = 10 # Add 10 users per step
spawn_rate = 5 # Spawn 5 users per second
time_limit = 300 # 5 minutes total
def tick(self):
run_time = self.get_run_time()
if run_time > self.time_limit:
return None # Stop test
current_step = run_time // self.step_time
user_count = int(current_step * self.step_load)
return (user_count, self.spawn_rate)
class WebUser(HttpUser):
wait_time = between(1, 3)
@task
def index(self):
self.client.get("/")Ramp Up/Down Pattern
class RampUpDownShape(LoadTestShape):
"""
Ramp up to peak load, maintain, then ramp down.
"""
ramp_up_time = 120 # 2 minutes ramp up
peak_time = 300 # 5 minutes at peak
ramp_down_time = 120 # 2 minutes ramp down
peak_users = 100 # Peak user count
spawn_rate = 10 # Users per second
def tick(self):
run_time = self.get_run_time()
total_time = self.ramp_up_time + self.peak_time + self.ramp_down_time
if run_time >= total_time:
return None
if run_time < self.ramp_up_time:
# Ramp up phase
progress = run_time / self.ramp_up_time
user_count = int(self.peak_users * progress)
elif run_time < self.ramp_up_time + self.peak_time:
# Peak phase
user_count = self.peak_users
else:
# Ramp down phase
ramp_down_start = self.ramp_up_time + self.peak_time
ramp_down_progress = (run_time - ramp_down_start) / self.ramp_down_time
user_count = int(self.peak_users * (1 - ramp_down_progress))
return (user_count, self.spawn_rate)Spike Testing Pattern
class SpikeTestShape(LoadTestShape):
"""
Spike testing with sudden load increases.
"""
base_users = 20
spike_users = 200
spike_duration = 60 # 1 minute spikes
normal_duration = 120 # 2 minutes normal
num_spikes = 3
spawn_rate = 20
def tick(self):
run_time = self.get_run_time()
cycle_time = self.spike_duration + self.normal_duration
total_time = cycle_time * self.num_spikes
if run_time >= total_time:
return None
cycle_position = run_time % cycle_time
if cycle_position < self.spike_duration:
# Spike phase
user_count = self.spike_users
else:
# Normal phase
user_count = self.base_users
return (user_count, self.spawn_rate)Time-Based Pattern
import math
class SineWaveShape(LoadTestShape):
"""
Sine wave load pattern for cyclical testing.
"""
min_users = 10
max_users = 100
period = 300 # 5 minute cycle
spawn_rate = 5
test_duration = 1800 # 30 minutes
def tick(self):
run_time = self.get_run_time()
if run_time > self.test_duration:
return None
# Calculate sine wave
amplitude = (self.max_users - self.min_users) / 2
mid_point = self.min_users + amplitude
wave_position = 2 * math.pi * run_time / self.period
user_count = int(mid_point + amplitude * math.sin(wave_position))
return (user_count, self.spawn_rate)User Class Distribution
class MultiUserShape(LoadTestShape):
"""
Shape with different user class distributions over time.
"""
def tick(self):
run_time = self.get_run_time()
if run_time > 600: # 10 minutes
return None
if run_time < 180: # First 3 minutes: only regular users
return (50, 10, [RegularUser])
elif run_time < 360: # Next 3 minutes: mixed users
return (75, 10, [RegularUser, PowerUser])
else: # Last 4 minutes: all user types
return (100, 10, [RegularUser, PowerUser, AdminUser])
class RegularUser(HttpUser):
weight = 3
wait_time = between(2, 5)
@task
def browse(self):
self.client.get("/")
class PowerUser(HttpUser):
weight = 2
wait_time = between(1, 3)
@task
def advanced_features(self):
self.client.get("/admin/dashboard")
class AdminUser(HttpUser):
weight = 1
wait_time = between(3, 8)
@task
def admin_tasks(self):
self.client.get("/admin/users")Conditional Load Patterns
class ConditionalShape(LoadTestShape):
"""
Load shape that adapts based on response times.
"""
target_response_time = 2000 # 2 seconds
max_users = 200
min_users = 10
adjustment_rate = 10
def tick(self):
run_time = self.get_run_time()
if run_time > 1800: # 30 minutes max
return None
current_users = self.get_current_user_count()
# Get average response time from stats
stats = self.runner.environment.stats
if stats.total.num_requests > 0:
avg_response_time = stats.total.avg_response_time
if avg_response_time > self.target_response_time:
# Response time too high, reduce users
new_users = max(self.min_users, current_users - self.adjustment_rate)
elif avg_response_time < self.target_response_time * 0.5:
# Response time good, increase users
new_users = min(self.max_users, current_users + self.adjustment_rate)
else:
# Response time acceptable, maintain
new_users = current_users
else:
# No stats yet, start with minimum
new_users = self.min_users
return (new_users, 5)Schedule-Based Pattern
import datetime
class ScheduleBasedShape(LoadTestShape):
"""
Load pattern based on business hours schedule.
"""
def get_business_hours_load(self, hour):
"""Get user count based on business hour."""
if 9 <= hour <= 17: # Business hours
if 12 <= hour <= 13: # Lunch hour peak
return 150
elif hour in [9, 17]: # Start/end of day
return 100
else: # Regular business hours
return 120
elif 18 <= hour <= 22: # Evening
return 60
else: # Night/early morning
return 20
def tick(self):
run_time = self.get_run_time()
if run_time > 86400: # 24 hours max
return None
# Simulate time-based load
current_hour = int((run_time / 3600) % 24)
user_count = self.get_business_hours_load(current_hour)
return (user_count, 10)Types
from typing import Optional, Tuple, List, Type, Union
from locust import User
# Shape return types
UserCount = int
SpawnRate = int
UserClasses = List[Type[User]]
# Shape tick return types
ShapeReturn = Union[
Tuple[UserCount, SpawnRate],
Tuple[UserCount, SpawnRate, UserClasses],
None # Stop test
]
# Load shape configuration
class LoadTestShape:
def tick(self) -> Optional[ShapeReturn]: ...
def reset_time(self) -> None: ...
def get_run_time(self) -> float: ...
def get_current_user_count(self) -> int: ...Install with Tessl CLI
npx tessl i tessl/pypi-locust