Action Executor

Overview

The action executor maps command strings to callable functions. Action scripts are JSON lists, so the same script can be hand-authored, generated by HAR import, scheduled by an MCP tool, or sent over the control socket.

Every shipped command starts with the LD_ prefix; safe Python built-ins (print, len, range…) are also available, but eval, exec, compile, __import__, breakpoint, open, and input are explicitly blocked.

Action format

["command_name"]                        # No parameters
["command_name", {"key": "value"}]      # Keyword arguments
["command_name", [arg1, arg2]]          # Positional arguments

The top-level document is either:

{"load_density": [["LD_start_test", {...}], ...]}

or a bare list of actions.

Quick example

from je_load_density import execute_action

execute_action({"load_density": [
    ["LD_register_variables", {"variables": {"base": "https://api.example.com"}}],
    ["LD_start_test", {
        "user_detail_dict": {"user": "fast_http_user"},
        "user_count": 20,
        "spawn_rate": 10,
        "test_time": 30,
        "tasks": [{"method": "get", "request_url": "${var.base}/health"}],
    }],
    ["LD_generate_summary_report", {"report_name": "smoke"}],
]})

LD_* commands

The executor exposes the following commands. Each is implemented in the matching module under je_load_density.

Core:

Command	Summary
`LD_start_test`	Run a Locust load test (HTTP / FastHttp / WebSocket / gRPC / MQTT / Socket).
`LD_execute_action`	Execute a nested action list.
`LD_execute_files`	Execute every action JSON file in a list.
`LD_add_package_to_executor`	Register a Python package’s functions into the executor.
`LD_start_socket_server`	Start the hardened TCP control plane.

Reports:

Command	Summary
`LD_generate_html` / `LD_generate_html_report`	HTML report generators.
`LD_generate_json` / `LD_generate_json_report`	JSON report generators.
`LD_generate_xml` / `LD_generate_xml_report`	XML report generators.
`LD_generate_csv_report`	One-row-per-request CSV export.
`LD_generate_junit_report`	JUnit XML for CI consumers.
`LD_generate_summary_report`	JSON summary with per-name p50/p90/p95/p99 latencies.
`LD_summary`	In-memory dict of the same summary.

Test record persistence:

Command	Summary
`LD_persist_records`	Save the in-memory records to a SQLite database.
`LD_list_runs`	List recent runs in a database.
`LD_fetch_run_records`	Load every record for one run.
`LD_clear_records`	Drop the in-memory record list.

Parameter resolver:

Command	Summary
`LD_register_variable` / `LD_register_variables`	Register one or many `${var.x}` values.
`LD_register_csv_source` / `LD_register_csv_sources`	Bind a CSV file to a `${csv.name.col}` source.
`LD_clear_resolver`	Reset every registered variable / source.

Recording / replay:

Command	Summary
`LD_load_har`	Read a HAR JSON file from disk.
`LD_har_to_tasks`	Convert a HAR document into a list of LoadDensity tasks.
`LD_har_to_action_json`	Convert a HAR document into a runnable action JSON.

Metrics exporters:

Command	Summary
`LD_start_prometheus_exporter` / `LD_stop_prometheus_exporter`	Toggle the Prometheus HTTP endpoint.
`LD_start_influxdb_sink` / `LD_stop_influxdb_sink`	Toggle the InfluxDB UDP / HTTP sink.
`LD_start_opentelemetry_exporter` / `LD_stop_opentelemetry_exporter`	Toggle the OTLP gRPC exporter.

Adding custom commands

from je_load_density import add_command_to_executor

def slack_notify(message: str) -> None:
    ...

add_command_to_executor({"LD_slack_notify": slack_notify})

Once registered, the new command is callable from any action JSON.