Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 141 additions & 0 deletions CLAUDE.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,141 @@
# LoadDensity

Load & Stress Automation Framework built on top of Locust.

## Tech Stack

- Python 3.10+
- Locust (load testing engine)
- PySide6 + qt-material (optional GUI)
- setuptools (build system)

## Project Structure

- `je_load_density/` - main package
- `gui/` - PySide6 GUI with multi-language support
- `utils/` - utilities (executor, file I/O, reports, logging, JSON/XML, socket server, test records)
- `wrapper/` - Locust wrappers (env creation, event hooks, proxy users, start/stop)
- `load_density_driver/` - driver generation
- `test/` - pytest test suite
- `docs/` - Sphinx documentation

## Development Commands

```bash
# Install
pip install -e .
pip install -e ".[gui]"

# Test
pytest test/

# Build
python -m build
```

## Coding Standards

### Design Patterns & Software Engineering

- Apply appropriate design patterns (Strategy, Factory, Observer, etc.) where they reduce complexity
- Follow SOLID principles: single responsibility, open-closed, Liskov substitution, interface segregation, dependency inversion
- Prefer composition over inheritance
- Keep functions small and focused on a single task
- Use meaningful, descriptive names for variables, functions, classes, and modules

### Performance

- Avoid unnecessary object creation in hot paths
- Prefer generators over lists for large data iteration
- Use appropriate data structures (set for membership checks, dict for lookups)
- Minimize I/O operations; batch when possible
- Profile before optimizing - measure, don't guess

### Code Hygiene

- Remove all unused imports, variables, functions, classes, and dead code blocks
- No commented-out code in commits
- No placeholder or stub code left behind
- Every import must be used; every function must be called or exported

### Security

- Never hardcode secrets, tokens, passwords, or API keys
- Validate and sanitize all external input (user input, file content, network data)
- Use parameterized queries for any database operations
- Avoid `eval()`, `exec()`, and `__import__()` with untrusted input
- Use `subprocess` with argument lists, never shell=True with user input
- Set restrictive file permissions on sensitive files
- Escape output to prevent injection (HTML, XML, JSON)
- Pin dependency versions to avoid supply chain attacks

### Linter Compliance (SonarQube / Codacy / Pylint / Flake8)

Code must pass static analysis with no new issues introduced. Follow these rules proactively so SonarQube, Codacy, Pylint, Flake8, Bandit, and Radon do not flag regressions.

#### Complexity & Structure

- Cognitive complexity per function: ≤ 15 (SonarQube rule `python:S3776`)
- Cyclomatic complexity per function: ≤ 10 (Radon grade A–B)
- Function length: ≤ 80 lines; file length: ≤ 1000 lines
- Max parameters per function: ≤ 7 (SonarQube `python:S107`)
- Max nesting depth: ≤ 4 levels (SonarQube `python:S134`)
- Avoid deeply nested `if/for/try` — extract helpers or use early returns
- No duplicated code blocks ≥ 10 lines (SonarQube `common-py:DuplicatedBlocks`)
- Keep boolean expressions simple: ≤ 3 operators (SonarQube `python:S1067`)

#### Naming & Style (PEP 8 + Pylint)

- `snake_case` for functions, methods, variables, modules
- `PascalCase` for classes; `UPPER_SNAKE_CASE` for module-level constants
- Private members prefixed with single underscore `_name`
- Line length: ≤ 120 characters (soft limit), hard max 160
- No single-letter names except loop counters (`i`, `j`, `k`) and comprehensions
- Avoid shadowing built-ins (`id`, `list`, `type`, `dict`, `file`, etc.)
- No unused function/method parameters — prefix with `_` if required by signature

#### Bug-Prone Patterns

- Never use mutable default arguments (`def f(x=[])`) — use `None` sentinel (SonarQube `python:S5644`)
- Do not compare with `==` / `!=` to `None`, `True`, `False` — use `is` / `is not`
- Do not catch bare `except:` — catch specific exceptions; never swallow silently
- Always re-raise with `raise` or `raise X from e`, preserving context
- Close resources with `with` context managers (files, sockets, locks)
- Do not modify a collection while iterating over it
- Avoid `assert` for runtime validation (stripped by `python -O`); raise explicit exceptions
- No `TODO` / `FIXME` / `XXX` comments without a tracked issue reference
- Remove unreachable code after `return`, `raise`, `break`, `continue`

#### Type Safety & API Design

- Public functions and methods should have type hints (parameters + return)
- Avoid `Any` unless truly dynamic; prefer `Optional[T]`, `Union[...]`, protocols
- Do not return inconsistent types from one function (e.g. `str` or `None` or `int`)
- Prefer `@dataclass` or `TypedDict` over ad-hoc dict payloads
- Use `enum.Enum` instead of string/int constants for closed sets

#### Security (Bandit + SonarQube Security Hotspots)

- No `eval`, `exec`, `pickle.loads`, `yaml.load` (use `yaml.safe_load`) on untrusted input
- No `hashlib.md5` / `sha1` for security purposes — use `sha256` or `blake2b`
- No `random` module for tokens/secrets — use `secrets` module
- No `tempfile.mktemp` — use `mkstemp` / `NamedTemporaryFile`
- Never log secrets, tokens, or raw request bodies containing credentials
- Validate file paths against traversal (`..`, absolute paths, symlinks)
- Set explicit timeouts on `requests.*` and socket operations

#### Testing Hygiene

- Tests must be deterministic — no reliance on wall-clock, network, or ordering
- Each test asserts something; no test without an `assert`
- Mock external side effects (filesystem writes, HTTP, subprocess)
- Test names describe behavior: `test_<unit>_<condition>_<expected>`

### Git Commit Rules

- Commit messages must NOT reference any AI tool, assistant, or model name
- No `Co-Authored-By` lines referencing AI
- Write commit messages as if authored solely by the developer
- Use conventional commit style: `feat:`, `fix:`, `refactor:`, `test:`, `docs:`, `chore:`
- Keep subject line under 72 characters
- Use imperative mood ("add feature" not "added feature")
18 changes: 11 additions & 7 deletions je_load_density/utils/executor/action_executor.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,11 @@
from je_load_density.utils.package_manager.package_manager_class import package_manager
from je_load_density.wrapper.start_wrapper.start_test import start_test

_UNSAFE_BUILTINS = frozenset({
"eval", "exec", "compile", "__import__",
"breakpoint", "open", "input",
})


class Executor:
"""
Expand Down Expand Up @@ -54,9 +59,11 @@ def __init__(self) -> None:
"LD_add_package_to_executor": package_manager.add_package_to_executor,
}

# 將所有 Python 內建函式加入事件字典
# Add all Python built-in functions to event_dict
# 將安全的 Python 內建函式加入事件字典,排除可執行任意程式碼者
# Add safe Python built-in functions; exclude those allowing arbitrary code execution
for name, func in getmembers(builtins, isbuiltin):
if name in _UNSAFE_BUILTINS:
continue
self.event_dict[name] = func

def _execute_event(self, action: list) -> Any:
Expand Down Expand Up @@ -100,11 +107,8 @@ def execute_action(self, action_list: Union[list, dict]) -> dict[str, Any]:

execute_record_dict: dict[str, Any] = {}

try:
if not isinstance(action_list, list) or len(action_list) == 0:
raise LoadDensityTestExecuteException(executor_list_error)
except Exception as error:
print(repr(error), file=sys.stderr)
if not isinstance(action_list, list) or len(action_list) == 0:
raise LoadDensityTestExecuteException(executor_list_error)

for action in action_list:
try:
Expand Down
46 changes: 24 additions & 22 deletions je_load_density/utils/generate_report/generate_html_report.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,15 @@
import sys
from threading import Lock
from html import escape
from typing import List, Tuple

from je_load_density.utils.exception.exceptions import LoadDensityHTMLException
from je_load_density.utils.exception.exception_tags import html_generate_no_data_tag
from je_load_density.utils.test_record.test_record_class import test_record_instance


def _safe(value: object) -> str:
return escape(str(value), quote=True)

# HTML 標頭 (HTML head)
_HTML_STRING_HEAD = """<!DOCTYPE html>
<html lang="en">
Expand Down Expand Up @@ -79,24 +83,24 @@ def generate_html() -> Tuple[List[str], List[str]]:

success_list: List[str] = [
_SUCCESS_TABLE.format(
Method=record.get("Method"),
test_url=record.get("test_url"),
name=record.get("name"),
status_code=record.get("status_code"),
text=record.get("text"),
content=record.get("content"),
headers=record.get("headers"),
Method=_safe(record.get("Method")),
test_url=_safe(record.get("test_url")),
name=_safe(record.get("name")),
status_code=_safe(record.get("status_code")),
text=_safe(record.get("text")),
content=_safe(record.get("content")),
headers=_safe(record.get("headers")),
)
for record in test_record_instance.test_record_list
]

failure_list: List[str] = [
_FAILURE_TABLE.format(
http_method=record.get("Method"),
test_url=record.get("test_url"),
name=record.get("name"),
status_code=record.get("status_code"),
error=record.get("error"),
http_method=_safe(record.get("Method")),
test_url=_safe(record.get("test_url")),
name=_safe(record.get("name")),
status_code=_safe(record.get("status_code")),
error=_safe(record.get("error")),
)
for record in test_record_instance.error_record_list
]
Expand All @@ -112,18 +116,16 @@ def generate_html_report(html_name: str = "default_name") -> str:
:param html_name: 輸出檔案名稱 (Output file name, without extension)
:return: HTML 字串 (HTML string)
"""
_lock = Lock()
success_list, failure_list = generate_html()

try:
with _lock: # 使用 with 確保自動 acquire/release
html_path = f"{html_name}.html"
with open(html_path, "w+", encoding="utf-8") as file_to_write:
file_to_write.write(_HTML_STRING_HEAD)
file_to_write.writelines(success_list)
file_to_write.writelines(failure_list)
file_to_write.write(_HTML_STRING_BOTTOM)
return html_path
html_path = f"{html_name}.html"
with open(html_path, "w+", encoding="utf-8") as file_to_write:
file_to_write.write(_HTML_STRING_HEAD)
file_to_write.writelines(success_list)
file_to_write.writelines(failure_list)
file_to_write.write(_HTML_STRING_BOTTOM)
return html_path
except Exception as error:
print(repr(error), file=sys.stderr)
return ""
5 changes: 3 additions & 2 deletions je_load_density/utils/generate_report/generate_json_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,8 @@
from je_load_density.utils.exception.exceptions import LoadDensityGenerateJsonReportException
from je_load_density.utils.test_record.test_record_class import test_record_instance

_json_report_lock = Lock()


def generate_json() -> Tuple[Dict[str, dict], Dict[str, dict]]:
"""
Expand Down Expand Up @@ -55,14 +57,13 @@ def generate_json_report(json_file_name: str = "default_name") -> Tuple[str, str
:param json_file_name: 輸出檔案名稱前綴 (Output file name prefix)
:return: (成功檔案路徑, 失敗檔案路徑)
"""
lock = Lock()
success_dict, failure_dict = generate_json()

success_path = f"{json_file_name}_success.json"
failure_path = f"{json_file_name}_failure.json"

try:
with lock: # 使用 with 確保自動 acquire/release
with _json_report_lock:
with open(success_path, "w+", encoding="utf-8") as file_to_write:
json.dump(success_dict, file_to_write, indent=4, ensure_ascii=False)

Expand Down
12 changes: 5 additions & 7 deletions je_load_density/utils/json/json_file/json_file.py
Original file line number Diff line number Diff line change
@@ -1,11 +1,13 @@
import json
from pathlib import Path
from threading import Lock
from typing import Any, Union
from typing import Union

from je_load_density.utils.exception.exceptions import LoadDensityTestJsonException
from je_load_density.utils.exception.exception_tags import cant_find_json_error, cant_save_json_error

_json_file_lock = Lock()


def read_action_json(json_file_path: str) -> Union[dict, list]:
"""
Expand All @@ -16,18 +18,15 @@ def read_action_json(json_file_path: str) -> Union[dict, list]:
:return: JSON 內容 (dict or list)
:raises LoadDensityTestJsonException: 當檔案不存在或無法讀取時 (if file not found or cannot be read)
"""
lock = Lock()
try:
with lock:
with _json_file_lock:
file_path = Path(json_file_path)
if file_path.exists() and file_path.is_file():
with open(json_file_path, "r", encoding="utf-8") as read_file:
return json.load(read_file)
else:
raise LoadDensityTestJsonException(cant_find_json_error)
except Exception as error:
# 捕捉所有錯誤並轉換成自訂例外
# Catch all errors and raise custom exception
raise LoadDensityTestJsonException(f"{cant_find_json_error}: {error}")


Expand All @@ -40,9 +39,8 @@ def write_action_json(json_save_path: str, action_json: Union[dict, list]) -> No
:param action_json: 要寫入的資料 (data to write, dict or list)
:raises LoadDensityTestJsonException: 當檔案無法寫入時 (if file cannot be saved)
"""
lock = Lock()
try:
with lock:
with _json_file_lock:
with open(json_save_path, "w+", encoding="utf-8") as file_to_write:
json.dump(action_json, file_to_write, indent=4, ensure_ascii=False)
except Exception as error:
Expand Down
2 changes: 1 addition & 1 deletion load_density_driver/generate_load_density_driver.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
from je_load_density.utils.socket_server.load_density_socket_server import start_load_density_socket_server

try:
server = start_load_density_socket_server()
start_load_density_socket_server()
except Exception as error:
print(repr(error))
Loading