蝗虫异常：没有定义任务使用@task装饰器或设置用户的任务属性
locust Exception: No tasks defined on use the @task decorator or set the tasks property of the User
源代码在这里https://github.com/anishnath/vault-load-testing/blob/master/locusts/key_value.py#L65
到运行
locust --host https://10.1.3.211:8200 --headless
您好专家遇到错误
: No tasks defined on KeyValueLocust. 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 2.8.2
密码
import json
import sys
import os
import random
from locust import HttpLocust, task
sys.path.append(os.getcwd())
import common
from locusts import VaultTaskSet, VaultLocust
class KeyValueTasks(VaultTaskSet):
def __init__(self, parent):
super().__init__(parent)
self.kv_version = 1
def on_start(self):
self.kv_version = common.get_kv_version(client=self.client)
@task
def get_kv_secret(self):
@task
def put_kv_secret(self):
@task
def list_l1_secrets(self):
@task
def list_l2_secrets(self):
key_path = common.key_path_1()
class KeyValueLocust(VaultLocust):
task_set = KeyValueTasks
weight = 3
min_wait = 5000
max_wait = 10000
locustfile.py
from locusts.key_value import KeyValueLocust
__static__ = [KeyValueLocust]
__all__ = __static__
from locust import TaskSet, HttpUser
from locust.clients import ResponseContextManager, HttpSession, RequestException, CatchResponseError
import time
import os
import json
class VaultLocust(HttpUser):
token = None
testdata = None
abstract = True
def __init__(self,parent):
super().__init__(parent)
self.client = VaultSession(base_url=self.host,request_event=self.client.request_event, user=self)
def setup(self):
@classmethod
def _set_token(cls, token):
cls.token = token
@classmethod
def _set_testdata(cls, data):
cls.testdata = data
class VaultTaskSet(TaskSet):
def mount(self, name: str, mount_point: str=None):
mount_point = mount_point or name
r = self.client.get('/v1/sys/mounts')
if f'{mount_point}/' not in r.json():
self.client.post(f'/v1/sys/mounts/{mount_point}', json={'type': name})
def enable_auth(self, name: str, path: str=None):
path = path or name
r = self.client.get('/v1/sys/auth')
if f'{path}/' not in r.json():
self.client.post(f'/v1/sys/auth/{path}', json={'type': name})
def revoke_lease(self, lease_id: str):
self.client.put('/v1/sys/leases/revoke',
json={'lease_id': lease_id})
def is_in_list(self, key: str, uri: str) -> bool:
with self.client.request('LIST', uri, catch_response=True) as r:
if r.status_code == 404:
r.success()
return False
else:
return key in r.json()['data']['keys']
@property
def client(self) -> HttpSession:
client = super().client # type: HttpSession
client.headers['X-Vault-Token'] = self.locust.token
return client
class VaultSession(HttpSession):
def request(self, method, url, name=None, catch_response=False, **kwargs):
# Load any TLS certificates specified in the VAULT_CACERT env var
self.verify = os.environ.get('VAULT_CACERT', None)
# prepend url with hostname unless it's already an absolute URL
url = self._build_url(url)
# store meta data that is used when reporting the request to locust's statistics
request_meta = dict()
# set up pre_request hook for attaching meta data to the request object
request_meta["method"] = method
request_meta["start_time"] = time.time()
response = self._send_request_safe_mode(method, url, **kwargs)
# record the consumed time
request_meta["response_time"] = int((time.time() - request_meta["start_time"]) * 1000)
request_meta["name"] = name or (response.history and response.history[0] or response).request.path_url
# get the length of the content, but if the argument stream is set to True, we take
# the size from the content-length header, in order to not trigger fetching of the body
if kwargs.get("stream", False):
request_meta["content_size"] = int(response.headers.get("content-length") or 0)
else:
request_meta["content_size"] = len(response.content or "")
if catch_response:
response.locust_request_meta = request_meta
return ResponseContextManager(response)
else:
try:
response.raise_for_status()
except RequestException as e:
try:
e = CatchResponseError('. '.join(response.json()['errors']))
except KeyError:
e = CatchResponseError(e)
except json.JSONDecodeError:
e = CatchResponseError(e)
events.request_failure.fire(
request_type=request_meta["method"],
name=request_meta["name"],
response_time=request_meta["response_time"],
exception=e,
)
else:
events.request_success.fire(
request_type=request_meta["method"],
name=request_meta["name"],
response_time=request_meta["response_time"],
response_length=request_meta["content_size"],
)
return response
====更新====
试过@cyberwiz 的方法仍然有同样的错误
class KeyValueLocust(VaultLocust):
task = [KeyValueTasks]
weight = 3
min_wait = 5000
max_wait = 10000
[2022-02-17 11:16:40,809] AV-MBP-709/ERROR/locust.user.task: No tasks defined on KeyValueLocust. 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)
Traceback (most recent call last):
File "/opt/PycharmProjects/vault-load-testing2/venv/lib/python3.9/site-packages/locust/user/task.py", line 304, in run
self.schedule_task(self.get_next_task())
File "/opt/PycharmProjects/vault-load-testing2/venv/lib/python3.9/site-packages/locust/user/task.py", line 436, in get_next_task
raise Exception(
Exception: No tasks defined on KeyValueLocust. 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)
回购：https://github.com/anishnath/vault-load-testing/blob/master/locusts/key_value.py#L65
您正在设置 task_set 而不是 tasks。
见http://docs.locust.io/en/stable/writing-a-locustfile.html#id2
源代码在这里https://github.com/anishnath/vault-load-testing/blob/master/locusts/key_value.py#L65
到运行
locust --host https://10.1.3.211:8200 --headless
您好专家遇到错误
: No tasks defined on KeyValueLocust. 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 2.8.2
密码
import json
import sys
import os
import random
from locust import HttpLocust, task
sys.path.append(os.getcwd())
import common
from locusts import VaultTaskSet, VaultLocust
class KeyValueTasks(VaultTaskSet):
def __init__(self, parent):
super().__init__(parent)
self.kv_version = 1
def on_start(self):
self.kv_version = common.get_kv_version(client=self.client)
@task
def get_kv_secret(self):
@task
def put_kv_secret(self):
@task
def list_l1_secrets(self):
@task
def list_l2_secrets(self):
key_path = common.key_path_1()
class KeyValueLocust(VaultLocust):
task_set = KeyValueTasks
weight = 3
min_wait = 5000
max_wait = 10000
locustfile.py
from locusts.key_value import KeyValueLocust
__static__ = [KeyValueLocust]
__all__ = __static__
from locust import TaskSet, HttpUser
from locust.clients import ResponseContextManager, HttpSession, RequestException, CatchResponseError
import time
import os
import json
class VaultLocust(HttpUser):
token = None
testdata = None
abstract = True
def __init__(self,parent):
super().__init__(parent)
self.client = VaultSession(base_url=self.host,request_event=self.client.request_event, user=self)
def setup(self):
@classmethod
def _set_token(cls, token):
cls.token = token
@classmethod
def _set_testdata(cls, data):
cls.testdata = data
class VaultTaskSet(TaskSet):
def mount(self, name: str, mount_point: str=None):
mount_point = mount_point or name
r = self.client.get('/v1/sys/mounts')
if f'{mount_point}/' not in r.json():
self.client.post(f'/v1/sys/mounts/{mount_point}', json={'type': name})
def enable_auth(self, name: str, path: str=None):
path = path or name
r = self.client.get('/v1/sys/auth')
if f'{path}/' not in r.json():
self.client.post(f'/v1/sys/auth/{path}', json={'type': name})
def revoke_lease(self, lease_id: str):
self.client.put('/v1/sys/leases/revoke',
json={'lease_id': lease_id})
def is_in_list(self, key: str, uri: str) -> bool:
with self.client.request('LIST', uri, catch_response=True) as r:
if r.status_code == 404:
r.success()
return False
else:
return key in r.json()['data']['keys']
@property
def client(self) -> HttpSession:
client = super().client # type: HttpSession
client.headers['X-Vault-Token'] = self.locust.token
return client
class VaultSession(HttpSession):
def request(self, method, url, name=None, catch_response=False, **kwargs):
# Load any TLS certificates specified in the VAULT_CACERT env var
self.verify = os.environ.get('VAULT_CACERT', None)
# prepend url with hostname unless it's already an absolute URL
url = self._build_url(url)
# store meta data that is used when reporting the request to locust's statistics
request_meta = dict()
# set up pre_request hook for attaching meta data to the request object
request_meta["method"] = method
request_meta["start_time"] = time.time()
response = self._send_request_safe_mode(method, url, **kwargs)
# record the consumed time
request_meta["response_time"] = int((time.time() - request_meta["start_time"]) * 1000)
request_meta["name"] = name or (response.history and response.history[0] or response).request.path_url
# get the length of the content, but if the argument stream is set to True, we take
# the size from the content-length header, in order to not trigger fetching of the body
if kwargs.get("stream", False):
request_meta["content_size"] = int(response.headers.get("content-length") or 0)
else:
request_meta["content_size"] = len(response.content or "")
if catch_response:
response.locust_request_meta = request_meta
return ResponseContextManager(response)
else:
try:
response.raise_for_status()
except RequestException as e:
try:
e = CatchResponseError('. '.join(response.json()['errors']))
except KeyError:
e = CatchResponseError(e)
except json.JSONDecodeError:
e = CatchResponseError(e)
events.request_failure.fire(
request_type=request_meta["method"],
name=request_meta["name"],
response_time=request_meta["response_time"],
exception=e,
)
else:
events.request_success.fire(
request_type=request_meta["method"],
name=request_meta["name"],
response_time=request_meta["response_time"],
response_length=request_meta["content_size"],
)
return response
====更新====
试过@cyberwiz 的方法仍然有同样的错误
class KeyValueLocust(VaultLocust):
task = [KeyValueTasks]
weight = 3
min_wait = 5000
max_wait = 10000
[2022-02-17 11:16:40,809] AV-MBP-709/ERROR/locust.user.task: No tasks defined on KeyValueLocust. 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)
Traceback (most recent call last):
File "/opt/PycharmProjects/vault-load-testing2/venv/lib/python3.9/site-packages/locust/user/task.py", line 304, in run
self.schedule_task(self.get_next_task())
File "/opt/PycharmProjects/vault-load-testing2/venv/lib/python3.9/site-packages/locust/user/task.py", line 436, in get_next_task
raise Exception(
Exception: No tasks defined on KeyValueLocust. 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)
回购：https://github.com/anishnath/vault-load-testing/blob/master/locusts/key_value.py#L65
您正在设置 task_set 而不是 tasks。
见http://docs.locust.io/en/stable/writing-a-locustfile.html#id2