回呼執行器
CallbackFunctionExecutor 可將觸發函式與回呼函式串連。
這對於測試後的工作流程非常有用 — 例如,執行測試後自動產生報告。
基本用法
from je_load_density import callback_executor
def after_test():
print("測試完成,正在產生報告...")
callback_executor.callback_function(
trigger_function_name="user_test",
callback_function=after_test,
user_detail_dict={"user": "fast_http_user"},
user_count=10,
spawn_rate=5,
test_time=5,
tasks={"get": {"request_url": "http://httpbin.org/get"}},
)
運作原理
在執行器的
event_dict中查找trigger_function_name使用提供的
**kwargs執行觸發函式觸發函式完成後,呼叫
callback_function回傳觸發函式的回傳值
可用的觸發函式
觸發名稱 |
函式 |
|---|---|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
傳遞參數給回呼函式
使用關鍵字參數(預設):
def my_callback(report_name, format_type):
print(f"正在產生 {format_type} 報告:{report_name}")
callback_executor.callback_function(
trigger_function_name="user_test",
callback_function=my_callback,
callback_function_param={"report_name": "final", "format_type": "html"},
callback_param_method="kwargs",
user_detail_dict={"user": "fast_http_user"},
user_count=10,
spawn_rate=5,
test_time=5,
tasks={"get": {"request_url": "http://httpbin.org/get"}},
)
使用位置參數:
def my_callback(arg1, arg2):
print(f"參數:{arg1}, {arg2}")
callback_executor.callback_function(
trigger_function_name="user_test",
callback_function=my_callback,
callback_function_param=["value1", "value2"],
callback_param_method="args",
user_detail_dict={"user": "fast_http_user"},
user_count=10,
spawn_rate=5,
test_time=5,
tasks={"get": {"request_url": "http://httpbin.org/get"}},
)
參數說明
參數 |
類型 |
說明 |
|---|---|---|
|
|
|
|
|
觸發後要執行的回呼函式 |
|
|
回呼函式的參數(dict 用於 kwargs,list 用於 args) |
|
|
|
|
— |
傳遞給觸發函式的參數 |
錯誤處理
以下情況會拋出
CallbackExecutorException:trigger_function_name不在event_dict中callback_param_method不是"kwargs"或"args"