fix: address SonarCloud quality issues on PR #105

Cognitive complexity (python:S3776) — extract helpers so each
function stays at or below 15:
- request_executor._check_assertions: split per assertion type into
  _ASSERTION_HANDLERS lookup with one helper per kind.
- scenario_runner._eval_condition: dispatch operators via the new
  _CONDITION_OPS table.
- mqtt/websocket _do_step: split into _dispatch_step plus per-method
  helpers (_publish, _subscribe, _send_only, _recv_only, etc.).
- har_importer._entry_to_task: factor out _extract_request_headers
  and _attach_post_body.
- socket_server.handle: factor out _handle_legacy_quit,
  _authorise_payload, and _dispatch_command.
- influxdb_sink.start_influxdb_sink: factor _validate_transport and
  _build_listener.

Other findings:
- python:S3516 on start_influxdb_sink and __main__.main: stop
  always-returning the same value (drop the redundant True from the
  sink, replace the unreachable raise in main with a print + exit
  code 2 so the return is honest).
- python:S4423 on the TLS server context: pin minimum_version to
  TLSv1_2 so older suites cannot be negotiated.
- python:S1192 on the response terminator: extract
  _RESPONSE_TERMINATOR (and _AUTH_FAILED sentinel) instead of
  duplicating the byte string.
- python:S1172 on execute_task/execute_tasks/run_scenario: drop the
  unused 'client' parameter; HTTP and FastHttp callers updated.
- python:S6353 on the dotted-path and function regexes: collapse
  [A-Za-z0-9_] to \w in parameter_resolver and grpc_user_template.
- python:S117 + python:S1481 in mcp_server: rename the Server local
  to server_cls in build_server, and drop the unused Server binding
  from run_stdio.

bandit clean on the full package; pytest test/ — 84 passed.

Author: JE-Chen

je_load_density/__main__.py

@@ -4,7 +4,6 @@
 from typing import List, Optional
 
 from je_load_density.utils.exception.exception_tags import argparse_get_wrong_data
-from je_load_density.utils.exception.exceptions import LoadDensityTestExecuteException
 from je_load_density.utils.executor.action_executor import execute_action, execute_files
 from je_load_density.utils.file_process.get_dir_file_list import get_dir_files_as_list
 from je_load_density.utils.json.json_file.json_file import read_action_json
@@ -114,7 +113,8 @@ def main(argv: Optional[List[str]] = None) -> int:
     if _dispatch_legacy(args):
         return 0
 
-    raise LoadDensityTestExecuteException(argparse_get_wrong_data)
+    print(argparse_get_wrong_data, file=sys.stderr)
+    return 2
 
 
 if __name__ == "__main__":

je_load_density/mcp_server/server.py

@@ -254,8 +254,8 @@ def build_server():
     """
     Build the MCP Server instance with the LoadDensity tool surface.
     """
-    Server, _, mcp_types = _ensure_mcp()
-    server = Server("loaddensity")
+    server_cls, _, mcp_types = _ensure_mcp()
+    server = server_cls("loaddensity")
 
     @server.list_tools()
     async def _list_tools():
@@ -283,7 +283,7 @@ def run_stdio() -> None:
     """
     Run the MCP server over stdio (the standard transport for Claude).
     """
-    Server, stdio_server, _ = _ensure_mcp()
+    _, stdio_server, _ = _ensure_mcp()
     server = build_server()
 
     import asyncio

je_load_density/utils/metrics/influxdb_sink.py

@@ -76,6 +76,46 @@ def _post_line_protocol(line: str, url: str, token: Optional[str], timeout: floa
         load_density_logger.warning(f"InfluxDB write failed: {error}")
 
 
+def _validate_transport(transport: str, url: Optional[str]) -> None:
+    if transport not in {"udp", "http"}:
+        raise ValueError(f"unsupported transport: {transport}")
+    if transport == "http":
+        if not url:
+            raise ValueError("url required when transport=http")
+        if not url.lower().startswith(_ALLOWED_URL_SCHEMES):
+            raise ValueError("InfluxDB URL must use http:// or https://")
+
+
+def _build_listener(config: Dict[str, Any]):
+    transport = config["transport"]
+    host = config["host"]
+    port = config["port"]
+    url = config["url"]
+    token = config["token"]
+    measurement = config["measurement"]
+    timeout = config["timeout"]
+
+    def _listener(request_type, name, response_time, response_length, exception=None, **_kwargs):
+        tags = {"request_type": str(request_type), "name": str(name)}
+        fields: Dict[str, Any] = {
+            "latency_ms": float(response_time or 0),
+            "response_bytes": int(response_length or 0),
+            "success": exception is None,
+        }
+        if exception is not None:
+            fields["error"] = repr(exception)[:512]
+        line = _build_line(measurement, tags, fields, time.time_ns())
+        try:
+            if transport == "udp":
+                _send_udp(line, host, port)
+            else:
+                _post_line_protocol(line, url, token, timeout)
+        except Exception as error:
+            load_density_logger.debug(f"InfluxDB write failed: {error}")
+
+    return _listener
+
+
 def start_influxdb_sink(
     transport: str = "udp",
     host: str = "127.0.0.1",
@@ -84,7 +124,7 @@ def start_influxdb_sink(
     token: Optional[str] = None,
     measurement: str = "loaddensity_request",
     timeout: float = 2.0,
-) -> bool:
+) -> None:
     """
     啟動 InfluxDB sink，將每個 request 寫入 line protocol。
     Start an InfluxDB sink that writes each request as a line-protocol
@@ -94,17 +134,11 @@ def start_influxdb_sink(
     InfluxDB endpoint; not a hard-coded destination.
     """
     transport = transport.lower()
-    if transport not in {"udp", "http"}:
-        raise ValueError(f"unsupported transport: {transport}")
-    if transport == "http":
-        if not url:
-            raise ValueError("url required when transport=http")
-        if not url.lower().startswith(_ALLOWED_URL_SCHEMES):
-            raise ValueError("InfluxDB URL must use http:// or https://")
+    _validate_transport(transport, url)
 
     with _lock:
         if _state["started"]:
-            return True
+            return
 
         config = {
             "transport": transport,
@@ -115,31 +149,13 @@ def start_influxdb_sink(
             "measurement": measurement,
             "timeout": timeout,
         }
+        listener = _build_listener(config)
 
-        def _listener(request_type, name, response_time, response_length, exception=None, **_kwargs):
-            tags = {"request_type": str(request_type), "name": str(name)}
-            fields: Dict[str, Any] = {
-                "latency_ms": float(response_time or 0),
-                "response_bytes": int(response_length or 0),
-                "success": exception is None,
-            }
-            if exception is not None:
-                fields["error"] = repr(exception)[:512]
-            line = _build_line(measurement, tags, fields, time.time_ns())
-            try:
-                if transport == "udp":
-                    _send_udp(line, host, port)
-                else:
-                    _post_line_protocol(line, url, token, timeout)
-            except Exception as error:
-                load_density_logger.debug(f"InfluxDB write failed: {error}")
-
-        events.request.add_listener(_listener)
+        events.request.add_listener(listener)
         _state["started"] = True
-        _state["listener"] = _listener
+        _state["listener"] = listener
         _state["config"] = config
         load_density_logger.info(f"InfluxDB sink started transport={transport}")
-        return True
 
 
 def stop_influxdb_sink() -> None:

je_load_density/utils/parameterization/parameter_resolver.py

@@ -6,7 +6,7 @@
 from typing import Any, Dict, Iterable, Iterator, List, Optional
 
 _PLACEHOLDER_PATTERN = re.compile(r"\$\{([^}]+)\}")
-_FUNCTION_PATTERN = re.compile(r"^([a-zA-Z_][a-zA-Z0-9_]*)\((.*)\)$")
+_FUNCTION_PATTERN = re.compile(r"^([a-zA-Z_]\w*)\((.*)\)$")
 
 
 class ParameterResolver:

je_load_density/utils/recording/har_importer.py

@@ -17,29 +17,17 @@ def load_har(file_path: str) -> Dict[str, Any]:
         return json.load(fh)
 
 
-def _entry_to_task(entry: Dict[str, Any]) -> Optional[Dict[str, Any]]:
-    request = entry.get("request") or {}
-    method = str(request.get("method", "")).lower()
-    url = request.get("url")
-    if not method or not url:
-        return None
-
-    headers = {}
-    for header in request.get("headers") or []:
+def _extract_request_headers(raw_headers: Any) -> Dict[str, str]:
+    headers: Dict[str, str] = {}
+    for header in raw_headers or []:
         name = str(header.get("name", "")).strip()
         value = header.get("value", "")
         if name and name.lower() not in _NON_REQUEST_HEADERS:
             headers[name] = value
+    return headers
 
-    task: Dict[str, Any] = {
-        "method": method,
-        "request_url": url,
-        "name": f"{method.upper()} {_path_only(url)}",
-    }
-    if headers:
-        task["headers"] = headers
 
-    post_data = request.get("postData") or {}
+def _attach_post_body(task: Dict[str, Any], post_data: Dict[str, Any]) -> None:
     mime = str(post_data.get("mimeType", "")).lower()
     text = post_data.get("text")
     params = post_data.get("params")
@@ -49,11 +37,32 @@ def _entry_to_task(entry: Dict[str, Any]) -> Optional[Dict[str, Any]]:
             task["json"] = json.loads(text)
         except json.JSONDecodeError:
             task["data"] = text
-    elif params:
+        return
+    if params:
         task["data"] = {p.get("name"): p.get("value") for p in params if p.get("name")}
-    elif text:
+        return
+    if text:
         task["data"] = text
 
+
+def _entry_to_task(entry: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+    request = entry.get("request") or {}
+    method = str(request.get("method", "")).lower()
+    url = request.get("url")
+    if not method or not url:
+        return None
+
+    task: Dict[str, Any] = {
+        "method": method,
+        "request_url": url,
+        "name": f"{method.upper()} {_path_only(url)}",
+    }
+    headers = _extract_request_headers(request.get("headers"))
+    if headers:
+        task["headers"] = headers
+
+    _attach_post_body(task, request.get("postData") or {})
+
     expected_status = (entry.get("response") or {}).get("status")
     if isinstance(expected_status, int) and expected_status:
         task["assertions"] = [{"type": "status_code", "value": expected_status}]

je_load_density/utils/socket_server/load_density_socket_server.py

@@ -15,6 +15,8 @@
 
 _MAX_PAYLOAD_BYTES = 1 << 20  # 1 MiB
 _FRAME_HEADER = struct.Struct("!I")
+_RESPONSE_TERMINATOR = b"Return_Data_Over_JE\n"
+_AUTH_FAILED = object()
 
 
 class TCPServer:
@@ -49,6 +51,9 @@ def __init__(
         self._tls_context: Optional[ssl.SSLContext] = None
         if certfile and keyfile:
             self._tls_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+            # Pin minimum TLS version so older insecure suites cannot be
+            # negotiated; PROTOCOL_TLS_SERVER alone permits TLS 1.0/1.1.
+            self._tls_context.minimum_version = ssl.TLSVersion.TLSv1_2
             self._tls_context.load_cert_chain(certfile=certfile, keyfile=keyfile)
 
     def socket_server(self, host: str, port: int) -> None:
@@ -119,49 +124,66 @@ def handle(self, connection) -> None:
             print(f"Command received: {len(command_string)} bytes", flush=True)
 
             if command_string == "quit_server":
-                if self.token is not None:
-                    self._send_frame(connection, b"Error: token required\n")
-                    return
-                self.close_flag = True
-                self._send_frame(connection, b"Server shutting down\n")
-                print("Now quit server", flush=True)
-                return
-
-            try:
-                payload = json.loads(command_string)
-            except json.JSONDecodeError as error:
-                self._send_frame(connection, f"Error: {error}\n".encode("utf-8"))
-                self._send_frame(connection, b"Return_Data_Over_JE\n")
+                self._handle_legacy_quit(connection)
                 return
 
-            command = payload
-            if isinstance(payload, dict) and ("token" in payload or "command" in payload):
-                if not self._check_token(payload.get("token")):
-                    self._send_frame(connection, b"Error: unauthorised\n")
-                    return
-                command = payload.get("command")
-                if payload.get("op") == "quit":
-                    self.close_flag = True
-                    self._send_frame(connection, b"Server shutting down\n")
-                    return
-            elif self.token is not None:
-                self._send_frame(connection, b"Error: token required\n")
+            command = self._authorise_payload(connection, command_string)
+            if command is _AUTH_FAILED:
                 return
-
             if command is None:
-                self._send_frame(connection, b"Return_Data_Over_JE\n")
+                self._send_frame(connection, _RESPONSE_TERMINATOR)
                 return
 
-            try:
-                for execute_return in execute_action(command).values():
-                    self._send_frame(connection, f"{execute_return}\n".encode("utf-8"))
-                self._send_frame(connection, b"Return_Data_Over_JE\n")
-            except Exception as error:
-                self._send_frame(connection, f"Error: {error}\n".encode("utf-8"))
-                self._send_frame(connection, b"Return_Data_Over_JE\n")
+            self._dispatch_command(connection, command)
         finally:
             connection.close()
 
+    def _handle_legacy_quit(self, connection) -> None:
+        if self.token is not None:
+            self._send_frame(connection, b"Error: token required\n")
+            return
+        self.close_flag = True
+        self._send_frame(connection, b"Server shutting down\n")
+        print("Now quit server", flush=True)
+
+    def _authorise_payload(self, connection, command_string: str):
+        """
+        Decode the JSON envelope, enforce the token, and return the
+        actual command to execute. Returns ``_AUTH_FAILED`` when the
+        client has already been answered (bad JSON, missing/bad token,
+        or a quit op was honoured).
+        """
+        try:
+            payload = json.loads(command_string)
+        except json.JSONDecodeError as error:
+            self._send_frame(connection, f"Error: {error}\n".encode("utf-8"))
+            self._send_frame(connection, _RESPONSE_TERMINATOR)
+            return _AUTH_FAILED
+
+        if isinstance(payload, dict) and ("token" in payload or "command" in payload):
+            if not self._check_token(payload.get("token")):
+                self._send_frame(connection, b"Error: unauthorised\n")
+                return _AUTH_FAILED
+            if payload.get("op") == "quit":
+                self.close_flag = True
+                self._send_frame(connection, b"Server shutting down\n")
+                return _AUTH_FAILED
+            return payload.get("command")
+
+        if self.token is not None:
+            self._send_frame(connection, b"Error: token required\n")
+            return _AUTH_FAILED
+
+        return payload
+
+    def _dispatch_command(self, connection, command) -> None:
+        try:
+            for execute_return in execute_action(command).values():
+                self._send_frame(connection, f"{execute_return}\n".encode("utf-8"))
+        except Exception as error:
+            self._send_frame(connection, f"Error: {error}\n".encode("utf-8"))
+        self._send_frame(connection, _RESPONSE_TERMINATOR)
+
 
 def start_load_density_socket_server(
     host: str = "localhost",

je_load_density/wrapper/user_template/fast_http_user_template.py

@@ -50,4 +50,4 @@ def test(self) -> None:
         proxy_user = locust_wrapper_proxy.user_dict.get("fast_http_user")
         if not proxy_user or not proxy_user.tasks:
             return
-        run_scenario(self.client, self.method, proxy_user.tasks)
+        run_scenario(self.method, proxy_user.tasks)

je_load_density/wrapper/user_template/grpc_user_template.py

@@ -28,7 +28,7 @@ def set_wrapper_grpc_user(user_detail_dict: Dict[str, Any], **kwargs) -> type:
     return GrpcUserWrapper
 
 
-_SAFE_DOTTED_PATH = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*(\.[A-Za-z_][A-Za-z0-9_]*)*$")
+_SAFE_DOTTED_PATH = re.compile(r"^[A-Za-z_]\w*(\.[A-Za-z_]\w*)*$")
 
 
 def _import_dotted(path: str) -> Any:

je_load_density/wrapper/user_template/http_user_template.py

@@ -50,4 +50,4 @@ def test(self) -> None:
         proxy_user = locust_wrapper_proxy.user_dict.get("http_user")
         if not proxy_user or not proxy_user.tasks:
             return
-        run_scenario(self.client, self.method, proxy_user.tasks)
+        run_scenario(self.method, proxy_user.tasks)