Skip to content

callback_doc.rst

Latest commit

 

History

History
133 lines (107 loc) · 3.75 KB

callback_doc.rst

File metadata and controls

133 lines (107 loc) · 3.75 KB

Callback Executor

The CallbackFunctionExecutor allows chaining a trigger function with a callback function. This is useful for post-test workflows — for example, run a test, then automatically generate a report.

Basic Usage

from je_load_density import callback_executor

def after_test():
    print("Test finished, generating report...")

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"}},
)

How It Works

  1. The trigger_function_name is looked up in the executor's event_dict
  2. The trigger function is executed with the provided **kwargs
  3. After the trigger function completes, the callback_function is called
  4. The return value of the trigger function is returned

Available Trigger Functions

Trigger Name Function
user_test start_test() — Run a load test
LD_generate_html generate_html() — Generate HTML fragments
LD_generate_html_report generate_html_report() — Generate HTML report file
LD_generate_json generate_json() — Generate JSON data
LD_generate_json_report generate_json_report() — Generate JSON report files
LD_generate_xml generate_xml() — Generate XML strings
LD_generate_xml_report generate_xml_report() — Generate XML report files

Passing Parameters to Callbacks

With keyword arguments (default):

def my_callback(report_name, format_type):
    print(f"Generating {format_type} report: {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"}},
)

With positional arguments:

def my_callback(arg1, arg2):
    print(f"Args: {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"}},
)

Parameters

Parameter Type Description
trigger_function_name str Name of function in event_dict to trigger
callback_function Callable Callback function to execute after the trigger
callback_function_param dict or list or None Parameters for the callback (dict for kwargs, list for args)
callback_param_method str "kwargs" (default) or "args"
**kwargs — Parameters passed to the trigger function

Error Handling

  • CallbackExecutorException is raised if:
    • trigger_function_name is not found in event_dict
    • callback_param_method is not "kwargs" or "args"