je_load_density

class je_load_density.AdaptiveRetryPolicy(transient_budget: int = 5, flaky_budget: int = 2, permanent_budget: int = 0, base_delay: float = 0.1, max_delay: float = 5.0, backoff_factor: float = 2.0, jitter: float = 0.25, classifier: Callable[[BaseException], str] = <function classify_error at 0x711192324860>, _counts: Dict[str, int] = <factory>)

Bases: object

backoff_factor: float = 2.0

base_delay: float = 0.1

budget_for(classification: str) → int

classifier() → str

decide(error: BaseException, attempt: int) → RetryDecision

flaky_budget: int = 2

jitter: float = 0.25

max_delay: float = 5.0

permanent_budget: int = 0

reset() → None

transient_budget: int = 5

exception je_load_density.CircuitOpenError

Bases: RuntimeError

Raised when the failure budget trips.

class je_load_density.FailureBudget(threshold: float = 0.05, window_seconds: float = 30.0, min_samples: int = 50, _events: Deque[tuple] = <factory>, _lock: <built-in function allocate_lock> = <factory>, _tripped: bool = False)

Bases: object

failure_rate(now: float | None = None) → float

is_breached(now: float | None = None) → bool

min_samples: int = 50

record(failed: bool, now: float | None = None) → None

sample_count() → int

threshold: float = 0.05

trip() → None

property tripped: bool

window_seconds: float = 30.0

class je_load_density.NetworkConditioner(latency_ms: float = 0.0, jitter_ms: float = 0.0, loss_rate: float = 0.0, name_filter: str | None = None)

Bases: object

applies_to(task: Dict[str, Any]) → bool

apply(task: ~typing.Dict[str, ~typing.Any], sleeper: ~typing.Callable[[float], None] = <built-in function sleep>) → None

jitter_ms: float = 0.0

latency_ms: float = 0.0

loss_rate: float = 0.0

name_filter: str | None = None

class je_load_density.OAuth2Client(token_url: str, client_id: str, client_secret: str, scope: Optional[str] = None, timeout: float = 5.0, safety_window: float = 30.0, poster: Callable[[str, bytes, Dict[str, str], float], Dict[str, Any]] = <function _default_poster at 0x711191e8bf60>, _cache: Dict[Tuple[str, str], je_load_density.utils.auth.oauth2._CacheEntry] = <factory>, _lock: <built-in function allocate_lock> = <factory>)

Bases: object

clear() → None

client_id: str

client_secret: str

get_client_credentials() → Dict[str, Any]

get_password(username: str, password: str) → Dict[str, Any]

poster(body: bytes, headers: Dict[str, str], timeout: float) → Dict[str, Any]

refresh(refresh: str) → Dict[str, Any]

safety_window: float = 30.0

scope: str | None = None

timeout: float = 5.0

token_url: str

class je_load_density.ProcessSupervisor(name_substrings: tuple = ('locust', 'gevent'), grace_seconds: float = 2.0, killed: List[int] = <factory>)

Bases: object

grace_seconds: float = 2.0

kill_orphans() → List[int]

killed: List[int]

name_substrings: tuple = ('locust', 'gevent')

class je_load_density.RpsThrottle(rps: float, burst: int | None = None)

Bases: object

Token-bucket throttle. acquire blocks until one token is available, then consumes it.

acquire() → None

je_load_density.SoakShape(**config)

je_load_density.SpikeShape(**config)

je_load_density.StagesShape(stages: List[Dict[str, Any]])

je_load_density.action_json_schema(known_commands: Iterable[str] | None = None) → Dict[str, Any]

Build the JSON Schema dict describing an action JSON document.

je_load_density.add_command_to_executor(command_dict: dict[str, Any]) → None

je_load_density.assert_sla(rules: Iterable[Dict[str, Any]], summary: Dict[str, Any] | None = None) → List[Dict[str, Any]]

Evaluate rules and raise LoadDensityAssertException on any failure. Returns the full result list on success.

je_load_density.build_load_shape(name: str, config: Dict[str, Any] | None = None)

Construct one of the built-in shapes by name. Returns the Locust LoadTestShape subclass, ready to pass to Environment(shape_class=…).

je_load_density.build_slack_summary(summary: Dict[str, Any], title: str = 'LoadDensity run') → Dict[str, Any]

je_load_density.build_summary() → Dict[str, Any]

彙整成功與失敗紀錄為統計摘要。 Build a summary dict of success/failure counts and per-name latency percentiles for charting and regression checks.

je_load_density.build_teams_summary(summary: Dict[str, Any], title: str = 'LoadDensity run') → Dict[str, Any]

je_load_density.classify_error(error: BaseException) → str

je_load_density.create_env(user_class: List[locust.User], another_event: locust.events = locust.events, runner_mode: str = 'local', master_bind_host: str = '*', master_bind_port: int = 5557, master_host: str = '127.0.0.1', master_port: int = 5557, load_shape: str | None = None, shape_config: Dict[str, Any] | None = None)

建立 Locust Environment 並依模式建立 runner。 Create Locust Environment and build the matching runner.

je_load_density.create_project_dir(project_path: str | None = None, parent_name: str = 'LoadDensity') → None

建立專案目錄結構 (Create project directory structure)

Parameters:

• project_path – 專案路徑 (Project path), 預設為當前工作目錄 (default: current working directory)

• parent_name – 專案主目錄名稱 (Project parent folder name)

je_load_density.curl_to_task(command: str) → Dict[str, Any]

Parse one cURL command into a LoadDensity HTTP task dict.

je_load_density.decode_jwt(token: str) → Tuple[Dict[str, Any], Dict[str, Any], bytes]

Decode the three JWT segments without verifying the signature. Returns (header_dict, payload_dict, signature_bytes).

je_load_density.diff_runs(database_path: str, baseline_run_id: int, current_run_id: int, tolerance: float = 0.1) → Dict[str, Any]

Compare two runs in database_path. tolerance is the fractional increase allowed (e.g. 0.10 = 10 %). Metrics that breach the tolerance are listed under regressions.

je_load_density.emit_github_annotations(title: str = 'LoadDensity', stream: TextIO | None = None) → int

Print one ::error:: annotation per failure record. Returns the number of annotations written.

je_load_density.evaluate_sla(rules: Iterable[Dict[str, Any]], summary: Dict[str, Any] | None = None) → List[Dict[str, Any]]

Evaluate every rule against the supplied summary (or build_summary() when omitted). Returns a list of result dicts.

je_load_density.execute_action(action_list: list) → dict[str, Any]

je_load_density.execute_files(execute_files_list: list[str]) → list[dict[str, Any]]

je_load_density.export_schema(path: str, known_commands: Iterable[str] | None = None) → str

Write the schema to disk. Returns the absolute path written.

je_load_density.extract_field(payload: Dict[str, Any], dotted_path: str) → Any

Pluck a value out of a parsed GraphQL response by dotted path (e.g. data.user.id). Returns None on miss.

je_load_density.fetch_client_credentials_token(token_url: str, client_id: str, client_secret: str, scope: str | None = None, timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, ~typing.Dict[str, str], float], ~typing.Dict[str, ~typing.Any]] = <function _default_poster>) → Dict[str, Any]

je_load_density.fetch_password_token(token_url: str, client_id: str, client_secret: str, username: str, password: str, scope: str | None = None, timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, ~typing.Dict[str, str], float], ~typing.Dict[str, ~typing.Any]] = <function _default_poster>) → Dict[str, Any]

je_load_density.fetch_run_records(database_path: str, run_id: int) → Iterable[Dict[str, Any]]

je_load_density.format_github_annotation(severity: str, message: str, file: str | None = None, line: int | None = None, title: str | None = None) → str

Build one ::<severity> [opts]::message line.

severity must be error, warning, or notice.

je_load_density.generate_chart_report(report_name: str = 'loaddensity-charts', bucket_size_seconds: float = 1.0) → Dict[str, str]

Render latency + RPS charts. Writes <report_name>-latency.png and <report_name>-rps.png and returns {"latency": path, "rps": path}.

je_load_density.generate_csv_report(csv_name: str = 'default_name') → str | None

產生 CSV 報告。 Generate a CSV report containing both success and failure records. Returns the path written, or None on failure.

je_load_density.generate_html() → Tuple[List[str], List[str]]

產生 HTML 片段 (Generate HTML fragments)

Returns:

(成功測試清單, 失敗測試清單) (list of success test HTML fragments, list of failure test HTML fragments)

je_load_density.generate_html_report(html_name: str = 'default_name') → str

產生完整 HTML 報告並輸出檔案 Generate full HTML report and save to file

Parameters:

html_name – 輸出檔案名稱 (Output file name, without extension)

Returns:

HTML 字串 (HTML string)

je_load_density.generate_json() → Tuple[Dict[str, dict], Dict[str, dict]]

產生測試紀錄的 JSON 結構 Generate JSON structure for test records

Returns:

(成功測試字典, 失敗測試字典) (success_dict, failure_dict)

je_load_density.generate_json_report(json_file_name: str = 'default_name') → Tuple[str, str]

輸出測試紀錄 JSON 報告 Generate JSON report files for test records

Parameters:

json_file_name – 輸出檔案名稱前綴 (Output file name prefix)

Returns:

(成功檔案路徑, 失敗檔案路徑)

je_load_density.generate_junit_report(report_name: str = 'loaddensity-junit') → str | None

產生 JUnit XML 報告，供 CI 系統消費。 Generate a JUnit XML report consumable by CI systems.

je_load_density.generate_summary_report(report_name: str = 'loaddensity-summary') → str | None

je_load_density.generate_xml() → Tuple[str, str]

產生 XML 字串 (Generate XML strings)

Returns:

(成功測試 XML 字串, 失敗測試 XML 字串) (success_xml_str, failure_xml_str)

je_load_density.generate_xml_report(xml_file_name: str = 'default_name') → Tuple[str, str]

輸出 XML 報告檔案 (Generate XML report files)

Parameters:

xml_file_name – 輸出檔案名稱前綴 (Output file name prefix)

Returns:

(成功檔案路徑, 失敗檔案路徑)

je_load_density.get_dir_files_as_list(dir_path: str = '/home/docs/checkouts/readthedocs.org/user_builds/loaddensity/checkouts/latest/docs/source', default_search_file_extension: str = '.json') → List[str]

取得指定目錄下所有符合副檔名的檔案清單 Get all files in a directory that end with the given extension

Parameters:

• dir_path – 要搜尋的目錄路徑 (Directory path to search)

• default_search_file_extension – 要搜尋的副檔名 (File extension to search, e.g. “.json”)

Returns:

檔案絕對路徑清單 (List of absolute file paths)

je_load_density.get_throttle(key: str, rps: float, burst: int | None = None) → RpsThrottle

Return the shared throttle for key; create it on first call. Subsequent calls ignore later rps / burst arguments — clear with reset_throttles() to reconfigure.

je_load_density.graphql_to_http_task(endpoint: str, query: str, variables: Dict[str, Any] | None = None, operation_name: str | None = None, headers: Dict[str, str] | None = None, name: str | None = None, assertions: list | None = None, extract: list | None = None) → Dict[str, Any]

Build a POST task that submits a GraphQL operation.

The returned dict is a normal LoadDensity HTTP task, so all placeholders, assertions, and extractors work as usual.

je_load_density.har_to_action_json(har: Dict[str, Any], user: str = 'fast_http_user', user_count: int = 10, spawn_rate: int = 5, test_time: int = 60, include: List[str] | None = None, exclude: List[str] | None = None) → Dict[str, Any]

將 HAR 轉成 LoadDensity action JSON。 Convert a HAR document into a complete action JSON ready to feed into execute_action.

je_load_density.har_to_tasks(har: Dict[str, Any], include: List[str] | None = None, exclude: List[str] | None = None) → List[Dict[str, Any]]

將 HAR 轉成任務清單。 Convert a HAR document into a list of LoadDensity tasks.

je_load_density.install_failure_budget(threshold: float = 0.05, window_seconds: float = 30.0, min_samples: int = 50, runner_quit_callback: Callable[[], None] | None = None) → FailureBudget

Subscribe to Locust request events; trip the breaker when the sliding-window failure rate exceeds threshold. The supplied runner_quit_callback is invoked once on breach.

je_load_density.install_network_conditioner(latency_ms: float = 0.0, jitter_ms: float = 0.0, loss_rate: float = 0.0, name_filter: str | None = None) → NetworkConditioner

je_load_density.jmeter_to_action_json(root, user: str = 'fast_http_user', user_count: int = 10, spawn_rate: int = 5, test_time: int = 60) → Dict[str, Any]

je_load_density.jmeter_to_tasks(root) → List[Dict[str, Any]]

je_load_density.k6_script_to_action_json(source: str, user: str = 'fast_http_user', user_count: int = 10, spawn_rate: int = 5, test_time: int = 60) → Dict[str, Any]

je_load_density.k6_script_to_tasks(source: str) → List[Dict[str, Any]]

Walk a k6 script string and return one task per http.* call.

je_load_density.lint_action(actions: List[Any] | Dict[str, Any], known_commands: Set[str] | None = None) → List[Dict[str, str]]

Lint an action JSON. Returns a list of finding dicts.

je_load_density.lint_action_file(path: str, known_commands: Set[str] | None = None) → List[Dict[str, str]]

Read an action JSON file from disk and lint it.

je_load_density.list_runs(database_path: str, limit: int = 20) → List[Dict[str, Any]]

je_load_density.load_har(file_path: str) → Dict[str, Any]

讀取 HAR JSON 檔。 Read a HAR JSON file from disk.

je_load_density.load_jmeter_jmx(file_path: str) → Any

je_load_density.load_k6_script(file_path: str) → str

je_load_density.load_openapi(file_path: str) → Dict[str, Any]

Load an OpenAPI document. YAML support is opt-in (pyyaml soft-dep).

je_load_density.load_postman_collection(file_path: str) → Dict[str, Any]

je_load_density.openapi_to_action_json(spec: Dict[str, Any], user: str = 'fast_http_user', user_count: int = 10, spawn_rate: int = 5, test_time: int = 60) → Dict[str, Any]

je_load_density.openapi_to_tasks(spec: Dict[str, Any]) → List[Dict[str, Any]]

je_load_density.persist_records(database_path: str, label: str | None = None, metadata: Dict[str, Any] | None = None) → int

將目前的測試紀錄寫入 SQLite。 Persist the current test records into SQLite. Returns the run id.

je_load_density.post_slack_summary(webhook_url: str, summary: ~typing.Dict[str, ~typing.Any] | None = None, title: str = 'LoadDensity run', timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, float], int] = <function _default_poster>) → int

Build a Slack Block Kit message from summary (defaults to build_summary()) and POST it. Returns the HTTP status.

je_load_density.post_teams_summary(webhook_url: str, summary: ~typing.Dict[str, ~typing.Any] | None = None, title: str = 'LoadDensity run', timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, float], int] = <function _default_poster>) → int

je_load_density.postman_to_action_json(collection: Dict[str, Any], user: str = 'fast_http_user', user_count: int = 10, spawn_rate: int = 5, test_time: int = 60) → Dict[str, Any]

je_load_density.postman_to_tasks(collection: Dict[str, Any]) → List[Dict[str, Any]]

Walk the collection and return one task per request.

je_load_density.prepare_env(user_class: List[locust.User], user_count: int = 50, spawn_rate: int = 10, test_time: int | None = 60, web_ui_dict: Dict[str, Any] | None = None, runner_mode: str = 'local', load_shape: str | None = None, shape_config: Dict[str, Any] | None = None, **kwargs)

啟動 Locust 環境，支援 local / master / worker 三種模式。 Prepare a Locust environment in local, master, or worker mode.

Distributed-mode fields (master_bind_host / master_bind_port / master_host / master_port / expected_workers) are read from **kwargs so the signature stays within the project’s public-API parameter budget.

je_load_density.read_action_json(json_file_path: str) → dict | list

讀取 JSON 檔案並回傳內容 Read JSON file and return its content

Parameters:

json_file_path – JSON 檔案路徑 (path to JSON file)

Returns:

JSON 內容 (dict or list)

Raises:

LoadDensityTestJsonException – 當檔案不存在或無法讀取時 (if file not found or cannot be read)

je_load_density.refresh_token(token_url: str, client_id: str, client_secret: str, refresh: str, timeout: float = 5.0, poster: ~typing.Callable[[str, bytes, ~typing.Dict[str, str], float], ~typing.Dict[str, ~typing.Any]] = <function _default_poster>) → Dict[str, Any]

je_load_density.register_csv_source(name: str, file_path: str, cycle: bool = True) → None

je_load_density.register_csv_sources(sources: Iterable[Dict[str, Any]]) → None

je_load_density.register_db_source(name: str, connection_string: str, query: str, cycle: bool = True) → None

je_load_density.register_db_sources(sources: Iterable[Dict[str, Any]]) → None

je_load_density.register_variable(name: str, value: Any) → None

je_load_density.register_variables(variables: Dict[str, Any]) → None

je_load_density.reset_throttles() → None

je_load_density.resolve(value: Any) → Any

je_load_density.run_with_retry(fn: ~typing.Callable[[], object], policy: ~je_load_density.utils.reliability.adaptive_retry.AdaptiveRetryPolicy | None = None, sleeper: ~typing.Callable[[float], None] = <built-in function sleep>) → object

Run fn under policy. Returns fn’s result on success; re-raises the last exception once the budget is exhausted.

je_load_density.sign_aws_request(method: str, url: str, region: str, service: str, access_key: str, secret_key: str, body: bytes | str = b'', headers: Mapping[str, str] | None = None, session_token: str | None = None, now: datetime | None = None) → Dict[str, str]

Return a new headers dict carrying the SigV4 Authorization header, x-amz-date, optional x-amz-security-token, and x-amz-content-sha256.

je_load_density.sign_jwt(payload: Dict[str, Any], secret: str | bytes, algorithm: str = 'HS256', headers: Dict[str, Any] | None = None, expires_in_seconds: int | None = None) → str

Build and sign a JSON Web Token. Mutates a copy of payload — if expires_in_seconds is set, exp is added.

je_load_density.snapshot_metrics(window_seconds: float = 10.0) → Dict[str, Any]

Build a snapshot suitable for the dashboard. rps is the rolling request count over window_seconds from the in-memory record list.

je_load_density.start_dashboard(host: str = '127.0.0.1', port: int = 8765, refresh_seconds: float = 1.0, window_seconds: float = 10.0) → LiveDashboardServer

je_load_density.start_influxdb_sink(transport: str = 'udp', host: str = '127.0.0.1', port: int = 8089, url: str | None = None, token: str | None = None, measurement: str = 'loaddensity_request', timeout: float = 2.0) → None

啟動 InfluxDB sink，將每個 request 寫入 line protocol。 Start an InfluxDB sink that writes each request as a line-protocol point. transport=udp uses host:port; transport=http requires url.

URL is caller-supplied so this defers to the operator’s configured InfluxDB endpoint; not a hard-coded destination.

je_load_density.start_load_density_socket_server(host: str = 'localhost', port: int = 9940, framed: bool = False, token: str | None = None, certfile: str | None = None, keyfile: str | None = None) → TCPServer

啟動 LoadDensity TCP 伺服器 Start LoadDensity TCP server.

The token may also come from the LOAD_DENSITY_SOCKET_TOKEN environment variable so secrets are not embedded in callers.

je_load_density.start_opentelemetry_exporter(endpoint: str | None = None, service_name: str = 'loaddensity', export_interval_ms: int = 5000) → bool

啟動 OpenTelemetry OTLP 指標輸出。 Start an OpenTelemetry OTLP metrics exporter and attach a Locust request listener.

je_load_density.start_prometheus_exporter(port: int = 9646, addr: str = '127.0.0.1') → int | None

啟動 Prometheus 指標伺服器並掛上 Locust request 事件監聽。 Start the Prometheus exporter HTTP server and attach a request listener.

Defaults to binding on loopback. Pass addr="0.0.0.0" explicitly to expose the exporter to the network when running in a container or remote node.

Returns the port the exporter is listening on, or None if the optional dependency is not installed.

je_load_density.start_statsd_sink(host: str = '127.0.0.1', port: int = 8125, prefix: str = 'loaddensity') → _StatsdEmitter

Subscribe to Locust request events and emit DogStatsD packets over UDP. Idempotent — calling twice rebinds.

je_load_density.start_test(user_detail_dict: Dict[str, Any], user_count: int = 50, spawn_rate: int = 10, test_time: int | None = 60, web_ui_dict: Dict[str, Any] | None = None, runner_mode: str = 'local', load_shape: str | None = None, shape_config: Dict[str, Any] | None = None, **kwargs) → Dict[str, Any]

啟動壓力測試。Start load test.

runner_mode is "local" | "master" | "worker". Distributed-mode fields (master_bind_host / master_bind_port / master_host / master_port / expected_workers) are accepted via **kwargs so the signature stays under the public API parameter budget.

je_load_density.stop_dashboard() → None

je_load_density.stop_influxdb_sink() → None

je_load_density.stop_opentelemetry_exporter() → None

je_load_density.stop_prometheus_exporter() → None

從 Locust 事件移除監聽器（注意 prometheus_client 的 server 無法輕易停止）。 Detach the listener; prometheus_client’s server keeps running.

je_load_density.stop_statsd_sink() → None

je_load_density.summarise_records(records: Iterable[Dict[str, Any]]) → Dict[str, Dict[str, float]]

Aggregate raw records into {name: {count, failures, p50, p95, p99, mean}}.

je_load_density.uninstall_failure_budget() → None

je_load_density.uninstall_network_conditioner() → None

je_load_density.with_watchdog(callable_: Callable[[...], Any], *args: Any, timeout_seconds: float = 300.0, on_timeout: Callable[[], None] | None = None, **kwargs: Any) → Any

Run callable_ and raise TimeoutError if it exceeds timeout_seconds. The callable continues running in its daemon thread until the process exits.

Modules

action_lsp
gui
mcp_server
tools
utils
wrapper