tessl/pypi-locust
Developer-friendly load testing framework for HTTP and other protocols with distributed testing capabilities.
—
Pending
tasksets.mddocs/
TaskSets and Task Management
TaskSets provide a way to organize related tasks and model complex user behavior patterns. Locust supports random task execution, sequential execution, and Markov chain-based probabilistic transitions between tasks.
Capabilities
TaskSet Base Class
The fundamental class for grouping and organizing tasks with shared state and lifecycle management.
class TaskSet:
"""
Base class for organizing related tasks with shared state.
Attributes:
tasks (list | dict): Tasks to execute - functions, classes, or dict with weights
min_wait (float): Deprecated - minimum wait time
max_wait (float): Deprecated - maximum wait time
wait_function (callable): Function to determine wait time between tasks
user: Parent user instance
parent: Parent TaskSet or User
client: HTTP client from user (convenience access)
"""
def __init__(self, parent):
"""
Initialize TaskSet.
Args:
parent: Parent TaskSet or User instance
"""
def on_start(self):
"""Called when TaskSet starts executing."""
def on_stop(self):
"""Called when TaskSet stops executing."""
def run(self):
"""Main execution loop - runs tasks with wait times."""
def execute_task(self, task):
"""
Execute a single task.
Args:
task: Task function or class to execute
"""
def schedule_task(self):
"""Schedule the next task to be executed."""
def get_next_task(self):
"""
Get the next task to execute based on task weights.
Returns:
Task function or class to execute
"""
def wait_time(self):
"""
Calculate wait time before next task.
Returns:
float: Wait time in seconds
"""
def wait(self):
"""Wait before executing next task."""
def interrupt(self, reschedule=True):
"""
Interrupt current TaskSet execution.
Args:
reschedule (bool): Whether to reschedule the task
"""
@property
def user(self):
"""Get the root User instance."""
@property
def parent(self):
"""Get the parent TaskSet or User."""
@property
def client(self):
"""Get HTTP client from user (convenience property)."""SequentialTaskSet
TaskSet that executes tasks in the order they are defined rather than randomly selecting them.
class SequentialTaskSet(TaskSet):
"""
TaskSet that executes tasks in sequential order.
Tasks are executed in the order they appear in the tasks list,
cycling back to the first task after the last one completes.
"""
# Inherits all TaskSet methods and attributes
# Overrides task selection to be sequential rather than randomMarkovTaskSet
TaskSet that uses Markov chains for probabilistic task transitions, enabling complex user behavior modeling.
class MarkovTaskSet(TaskSet):
"""
TaskSet using Markov chains for task transitions.
Tasks transition to other tasks based on probabilities defined
using @transition and @transitions decorators.
Attributes:
current (str): Current task/state name
abstract (bool): Mark as abstract class
"""
# Use with @transition and @transitions decorators to define state transitionsTask Decorators
task Decorator
Decorator to mark methods as tasks and specify their execution weight.
def task(weight=1):
"""
Decorator to mark a method as a task.
Args:
weight (int): Relative weight for task selection (default: 1)
Higher weights make tasks more likely to be selected
Returns:
Decorated function marked as a task
Usage:
@task # weight=1
def simple_task(self): ...
@task(3) # weight=3 (3x more likely)
def important_task(self): ...
"""tag Decorator
Decorator to tag tasks for filtering during test execution.
def tag(*tags):
"""
Decorator to tag tasks for filtering.
Args:
*tags (str): Variable number of tag strings
Returns:
Decorated function with tags
Usage:
@tag("api", "critical")
@task
def api_test(self): ...
@tag("slow")
@task
def slow_operation(self): ...
"""Markov Task Decorators
transition Decorator
Decorator for defining single transitions in Markov TaskSets.
def transition(func_name: str, weight: int = 1):
"""
Define a single transition from current task to target task.
Args:
func_name (str): Name of target task function
weight (int): Transition weight (default: 1)
Returns:
Decorated function with transition
Usage:
@transition("task_b", weight=2)
@task
def task_a(self): ...
"""transitions Decorator
Decorator for defining multiple transitions in Markov TaskSets.
def transitions(weights):
"""
Define multiple transitions from current task to target tasks.
Args:
weights (dict | list): Dict of {func_name: weight} or
list of (func_name, weight) tuples or func_names
Returns:
Decorated function with transitions
Usage:
@transitions({"task_b": 2, "task_c": 1})
@task
def task_a(self): ...
@transitions([("task_b", 2), ("task_c", 1)])
@task
def task_a(self): ...
"""Usage Examples
Basic TaskSet Example
from locust import HttpUser, TaskSet, task, between
class UserBehavior(TaskSet):
def on_start(self):
# Setup for this TaskSet
self.login()
def login(self):
self.client.post("/login", json={
"username": "testuser",
"password": "secret"
})
@task(3)
def browse_pages(self):
pages = ["/", "/about", "/products", "/contact"]
page = random.choice(pages)
self.client.get(page)
@task(1)
def search(self):
query = random.choice(["python", "testing", "locust"])
self.client.get(f"/search?q={query}")
def on_stop(self):
# Cleanup for this TaskSet
self.client.post("/logout")
class WebsiteUser(HttpUser):
tasks = [UserBehavior]
wait_time = between(1, 3)Nested TaskSets Example
from locust import HttpUser, TaskSet, task, between
class AdminBehavior(TaskSet):
weight = 1 # Less likely to be selected
@task
def manage_users(self):
self.client.get("/admin/users")
@task
def view_reports(self):
self.client.get("/admin/reports")
class ShoppingBehavior(TaskSet):
weight = 3 # More likely to be selected
@task(2)
def browse_products(self):
self.client.get("/products")
@task(1)
def add_to_cart(self):
product_id = random.randint(1, 100)
self.client.post(f"/cart/add/{product_id}")
class WebsiteUser(HttpUser):
tasks = [AdminBehavior, ShoppingBehavior]
wait_time = between(1, 5)SequentialTaskSet Example
from locust import HttpUser, SequentialTaskSet, task, between
class OrderProcess(SequentialTaskSet):
"""Tasks execute in order: browse -> select -> checkout -> complete"""
@task
def browse_products(self):
self.client.get("/products")
@task
def select_product(self):
self.product_id = random.randint(1, 100)
self.client.get(f"/products/{self.product_id}")
@task
def add_to_cart(self):
self.client.post("/cart/add", json={
"product_id": self.product_id,
"quantity": 1
})
@task
def checkout(self):
self.client.get("/checkout")
self.client.post("/checkout", json={
"payment_method": "credit_card"
})
@task
def complete_order(self):
self.client.get("/order/complete")
# After completion, cycle back to browse_products
class ShoppingUser(HttpUser):
tasks = [OrderProcess]
wait_time = between(2, 5)MarkovTaskSet Example
from locust import HttpUser, MarkovTaskSet, task, between
from locust.user.markov_taskset import transition, transitions
class UserJourney(MarkovTaskSet):
"""User behavior modeled as Markov chain with probabilistic transitions"""
@task
@transitions({"browse": 3, "search": 1, "logout": 1})
def homepage(self):
"""Entry point - users can browse, search, or logout"""
self.client.get("/")
@task
@transitions({"product_page": 2, "homepage": 1})
def browse(self):
"""Browse products - can go to specific product or back to homepage"""
self.client.get("/products")
@task
@transition("product_page", weight=3)
@transition("homepage", weight=1)
def search(self):
"""Search for products"""
query = random.choice(["laptop", "phone", "tablet"])
self.client.get(f"/search?q={query}")
@task
@transitions([("add_to_cart", 2), ("browse", 2), ("homepage", 1)])
def product_page(self):
"""View product details"""
product_id = random.randint(1, 100)
self.client.get(f"/products/{product_id}")
@task
@transitions({"checkout": 3, "browse": 1})
def add_to_cart(self):
"""Add item to cart"""
self.client.post("/cart/add", json={"product_id": 123})
@task
@transition("logout", weight=1)
def checkout(self):
"""Complete purchase"""
self.client.post("/checkout")
def logout(self):
"""End session - no @task decorator, only reached via transitions"""
self.client.post("/logout")
self.interrupt() # End this TaskSet
class MarkovUser(HttpUser):
tasks = [UserJourney]
wait_time = between(1, 3)TaskSet with Shared State
from locust import HttpUser, TaskSet, task, between
class APITestSuite(TaskSet):
def on_start(self):
# Authenticate and store token
response = self.client.post("/auth/login", json={
"username": "api_user",
"password": "secret"
})
self.auth_token = response.json()["access_token"]
# Set headers for all subsequent requests
self.client.headers.update({
"Authorization": f"Bearer {self.auth_token}"
})
# Initialize shared test data
self.created_resources = []
@task(2)
def create_resource(self):
"""Create a new resource"""
resource_data = {
"name": f"test_resource_{random.randint(1000, 9999)}",
"description": "Test resource created by Locust"
}
response = self.client.post("/api/resources", json=resource_data)
if response.status_code == 201:
resource_id = response.json()["id"]
self.created_resources.append(resource_id)
@task(5)
def read_resource(self):
"""Read existing resources"""
if self.created_resources:
resource_id = random.choice(self.created_resources)
self.client.get(f"/api/resources/{resource_id}")
else:
# Fallback to reading first resource
self.client.get("/api/resources/1")
@task(1)
def update_resource(self):
"""Update an existing resource"""
if self.created_resources:
resource_id = random.choice(self.created_resources)
update_data = {
"description": f"Updated at {time.time()}"
}
self.client.put(f"/api/resources/{resource_id}", json=update_data)
@task(1)
def delete_resource(self):
"""Delete a resource"""
if self.created_resources:
resource_id = self.created_resources.pop()
self.client.delete(f"/api/resources/{resource_id}")
def on_stop(self):
# Cleanup remaining resources
for resource_id in self.created_resources:
self.client.delete(f"/api/resources/{resource_id}")
class APIUser(HttpUser):
tasks = [APITestSuite]
wait_time = between(0.5, 2)Types
from typing import Union, List, Dict, Callable, Any, Optional
# Task definition types
TaskFunction = Callable[[], None]
TaskClass = type # TaskSet subclass
TaskWeight = int
TaskDict = Dict[Union[TaskFunction, TaskClass], TaskWeight]
TaskList = List[Union[TaskFunction, TaskClass, tuple]]
# TaskSet configuration
Tasks = Union[TaskList, TaskDict]
WaitFunction = Callable[[], float]
# Markov chain types
TransitionWeights = Dict[str, int]
TransitionList = List[Union[tuple[str, int], str]]
Transitions = Union[TransitionWeights, TransitionList]Install with Tessl CLI
npx tessl i tessl/pypi-locust