以编程方式分配蝗虫任务集
以编程方式分配蝗虫任务集
提问于 2021-09-28 01:44:18
我试图根据通过UI传递的主机值动态分配蝗虫任务。在本例中，如果主机作为" hello“传入，则测试应该运行hello任务，否则它将运行world任务。
from locust import HttpUser, TaskSet, task, events
class RandomTask1(TaskSet):
@task(1)
def soemthing(self):
print("Hello!")
class RandomTask2(TaskSet):
@task(1)
def soemthing(self):
print("World!")
class LoadTestUser(HttpUser):
def on_start(self):
host_config = self.host
if host_config == "hello":
tasks = {RandomTask1:1}
else:
tasks = {RandomTask2:1}下面的示例不起作用，我得到以下错误
Exception: No tasks defined on LoadTestUser. use the @task decorator or set the tasks property of the User (or mark it as abstract = True if you only intend to subclass it)你知道我怎么能做到这样的事吗？我已经对示例进行了简化，但对于所有意图和目的，让我们假设locust实例已经在运行，不能停止或重新启动，并且需要动态分配任务。
编辑：
试过这样做：
class LoadTestUser(HttpUser):
def on_start(self):
if self.host == "hello":
self.tasks = {HelloTask: 1}
else:
self.tasks = {WorldTask: 1}
@task
def nothing(self):
pass
class HelloTask(TaskSet):
@task
def something(self):
print("Hello")
class WorldTask(TaskSet):
@task
def something(self):
print("World")现在我看到了以下错误：
Traceback (most recent call last):
File "/Users/user/project/venv/lib/python3.8/site-packages/locust/user/task.py", line 285, in run
self.schedule_task(self.get_next_task())
File "/Users/user/project/venv/lib/python3.8/site-packages/locust/user/task.py", line 420, in get_next_task
return random.choice(self.user.tasks)
File "/Users/user/opt/anaconda3/lib/python3.8/random.py", line 291, in choice
return seq[i]
KeyError: 0回答 2
Stack Overflow用户
回答已采纳
发布于 2021-09-28 03:52:20
创建一个单独的任务，并将逻辑放在该任务中，以便运行它。
class LoadTestUser(HttpUser):
def something1(self):
print("Hello!")
def something2(self):
print("World!")
@task
def task_logic(self):
if self.host == "hello":
self.something1()
else:
self.something2()但是，您可以直接处理所得到的错误。即使您打算用TaskSets覆盖或更改任务，也需要在类中定义一个任务。这里有一个示例在文件中，但是只需添加一个带有pass的任务，这样它就什么都不做了，那么您的重写就能工作了。
class LoadTestUser(HttpUser):
def on_start(self):
host_config = self.host
if host_config == "hello":
self.tasks = {RandomTask1:1}
else:
self.tasks = {RandomTask2:1}
@task
def nothing(self):
pass编辑：，这应该可以工作，但是看起来在当前版本的Locust中可能有一个bug，它只在Locust第一次启动时接受一个用于任务的字典，然后只接受一个列表。在它被修正之前，另一个答案中的例子是有效的。
Stack Overflow用户
发布于 2021-09-28 16:47:10
import sys
from locust import HttpUser, TaskSet, task, events
class LoadTestUser(HttpUser):
def locust_class(self, name):
module = sys.modules[__name__]
return getattr(module, f"{name.capitalize()}Task")
def get_weighted_tasks(self, task_list):
new_tasks = []
for item in task_list:
if "locust_task_weight" in dir(item):
for i in range(item.locust_task_weight):
new_tasks.append(item)
return new_tasks
def get_locust_tasks(self, cls):
tasks = []
for maybe_task in cls.__dict__.values():
if hasattr(maybe_task, "locust_task_weight"):
tasks.append(maybe_task)
return tasks
def on_start(self):
task_cls = self.locust_class(self.host)
task_list = self.get_locust_tasks(task_cls)
self.tasks = self.get_weighted_tasks(task_list)
@task(1)
def nothing(self):
pass
class HelloTask(TaskSet):
@task(1)
def something(self):
print("Hello")
@task(100)
def something_else(self):
print("hello")
class WorldTask(TaskSet):
@task(1)
def something(self):
print("World")
@task(10)
def something_else(self):
print("world")页面原文内容由Stack Overflow提供。腾讯云小微IT领域专用引擎提供翻译支持
原文链接：
https://stackoverflow.com/questions/69354870
复制相关文章
相似问题
腾讯云开发者
