I need to pass headers to a sequence of API calls (all of which extend both SequentialTaskSet
and ApiUser
; ApiUser
being an extension of HttpUser
(, the Locust documentation for TaskSet told me that nesting was the best way to do that. However, I run into type errors (marked with # ERROR:
). What’s the best way to fix them?
What the ApiUser
looks like (the post_headers
are needed for authentication):
from locust import HttpUser
class ApiUser(HttpUser):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.users = None
self.access_token = None
self.refresh_token = None
self.headers = None
self.ref_data = ref.RefData()
self.post_headers = None
def post_helper(self, url: str, json: dict):
"""do something with error handling"""
def get_helper(self, path: str):
"""do something with error handling"""
What the SequentialTaskSet
looks like (PrepTasks
will be called elsewhere, the problem is in PrepTasks
itself):
from api_user import ApiUser
from locust import SequentialTaskSet, task
from typing import Dict, Any
class PrepTasks(ApiUser):
tasks = [QuickStartSequence] # ERROR: not defined
def __init__(self, post_headers: Dict[str, Any], *args, **kwargs):
super().__init__(*args, **kwargs)
self.post_headers = post_headers
class QuickStartSequence(SequentialTaskSet):
tasks = [Task1, Task2, Task3] # ERROR: not defined
class Task1(SequentialTaskSet, ApiUser):
@task
def get_call1(self):
return self.get_helper("some-path/some-sub") # ApiUser.get_helper()
@task
def stop(self):
self.interrupt()
class Task2(SequentialTaskSet, ApiUser):
@task
def get_call2(self):
return self.get_helper("some-path/some-sub") # ApiUser.get_helper()
@task
def stop(self):
self.interrupt()
class Task3(SequentialTaskSet, ApiUser):
@task
def get_call3(self):
return self.get_helper("some-path/some-sub") # ApiUser.get_helper()
@task
def stop(self):
self.interrupt()