From bf7665345afa0d27c27fe00841df3ddab2f0c0aa Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Fri, 19 Jun 2026 11:38:19 +0800 Subject: [PATCH 1/2] Add CycloneDX SBOM generator and duration-aware suite sharding --- README.md | 8 ++ README/README_zh-CN.md | 8 ++ README/README_zh-TW.md | 8 ++ .../Eng/doc/new_features/v18_features_doc.rst | 55 +++++++++ docs/source/Eng/eng_index.rst | 1 + .../Zh/doc/new_features/v18_features_doc.rst | 50 +++++++++ docs/source/Zh/zh_index.rst | 1 + je_auto_control/__init__.py | 6 + .../gui/script_builder/command_schema.py | 27 +++++ .../utils/executor/action_executor.py | 29 +++++ .../utils/mcp_server/tools/_factories.py | 48 ++++++++ .../utils/mcp_server/tools/_handlers.py | 20 ++++ je_auto_control/utils/sbom/__init__.py | 4 + je_auto_control/utils/sbom/sbom.py | 104 ++++++++++++++++++ je_auto_control/utils/test_shard/__init__.py | 6 + .../utils/test_shard/test_shard.py | 77 +++++++++++++ test/unit_test/headless/test_ops_batch.py | 93 ++++++++++++++++ 17 files changed, 545 insertions(+) create mode 100644 docs/source/Eng/doc/new_features/v18_features_doc.rst create mode 100644 docs/source/Zh/doc/new_features/v18_features_doc.rst create mode 100644 je_auto_control/utils/sbom/__init__.py create mode 100644 je_auto_control/utils/sbom/sbom.py create mode 100644 je_auto_control/utils/test_shard/__init__.py create mode 100644 je_auto_control/utils/test_shard/test_shard.py create mode 100644 test/unit_test/headless/test_ops_batch.py diff --git a/README.md b/README.md index 2b71ebf1..063b9621 100644 --- a/README.md +++ b/README.md @@ -13,6 +13,7 @@ ## Table of Contents +- [What's new (2026-06-19) — SBOM & Suite Sharding](#whats-new-2026-06-19--sbom--suite-sharding) - [What's new (2026-06-19) — Reactive Observer](#whats-new-2026-06-19--reactive-observer) - [What's new (2026-06-19) — WCAG 2.2 Audit](#whats-new-2026-06-19--wcag-22-audit) - [What's new (2026-06-19) — Memory & Determinism](#whats-new-2026-06-19--memory--determinism) @@ -70,6 +71,13 @@ --- +## What's new (2026-06-19) — SBOM & Suite Sharding + +Two pure-stdlib ops tools (security + scale research angles), full stack. Full reference: [`docs/source/Eng/doc/new_features/v18_features_doc.rst`](docs/source/Eng/doc/new_features/v18_features_doc.rst). + +- **CycloneDX SBOM** — `build_sbom` / `write_sbom` (`AC_generate_sbom`, `ac_generate_sbom`): emit a CycloneDX 1.6 dependency SBOM (name/version/purl/license) for supply-chain compliance (EU CRA / EO 14028); `root` limits to a package's closure, `extra_components` inventories action files. No third-party dependency. +- **Duration-aware suite sharding** — `shard_flows` / `merge_results` (`AC_shard_suite` / `AC_merge_results`): bin-pack flows into N shards balanced by historical per-flow duration (so the slowest worker, not test count, defines runtime), then merge per-shard reports into one rollup. + ## What's new (2026-06-19) — Reactive Observer A non-blocking screen observer (SikuliX `observe` model), full stack (facade, `AC_*`, MCP, Script Builder). Full reference: [`docs/source/Eng/doc/new_features/v17_features_doc.rst`](docs/source/Eng/doc/new_features/v17_features_doc.rst). diff --git a/README/README_zh-CN.md b/README/README_zh-CN.md index f660ebd0..f0f3473e 100644 --- a/README/README_zh-CN.md +++ b/README/README_zh-CN.md @@ -12,6 +12,7 @@ ## 目录 +- [本次更新 (2026-06-19) — SBOM 与测试分片](#本次更新-2026-06-19--sbom-与测试分片) - [本次更新 (2026-06-19) — 反应式观察器](#本次更新-2026-06-19--反应式观察器) - [本次更新 (2026-06-19) — WCAG 2.2 审计](#本次更新-2026-06-19--wcag-22-审计) - [本次更新 (2026-06-19) — 记忆与确定性](#本次更新-2026-06-19--记忆与确定性) @@ -69,6 +70,13 @@ --- +## 本次更新 (2026-06-19) — SBOM 与测试分片 + +来自安全与规模研究角度的两项纯标准库运维工具,走完整五层。完整参考:[`docs/source/Zh/doc/new_features/v18_features_doc.rst`](../docs/source/Zh/doc/new_features/v18_features_doc.rst)。 + +- **CycloneDX SBOM** — `build_sbom` / `write_sbom`(`AC_generate_sbom`、`ac_generate_sbom`):为供应链合规(欧盟 CRA / EO 14028)输出 CycloneDX 1.6 依赖 SBOM(name/version/purl/许可证);`root` 限定某包的闭包,`extra_components` 可纳入 action 文件。无第三方依赖。 +- **时长感知套件分片** — `shard_flows` / `merge_results`(`AC_shard_suite` / `AC_merge_results`):依每个流程历史时长把流程装箱成 N 片(让最慢的 worker 而非测试数量决定总时长),再把各分片报告合并为一份汇总。 + ## 本次更新 (2026-06-19) — 反应式观察器 非阻塞的屏幕观察器(SikuliX `observe` 模型),走完整五层(facade、`AC_*`、MCP、Script Builder)。完整参考:[`docs/source/Zh/doc/new_features/v17_features_doc.rst`](../docs/source/Zh/doc/new_features/v17_features_doc.rst)。 diff --git a/README/README_zh-TW.md b/README/README_zh-TW.md index 8ff86117..a41fceda 100644 --- a/README/README_zh-TW.md +++ b/README/README_zh-TW.md @@ -12,6 +12,7 @@ ## 目錄 +- [本次更新 (2026-06-19) — SBOM 與測試分片](#本次更新-2026-06-19--sbom-與測試分片) - [本次更新 (2026-06-19) — 反應式觀察器](#本次更新-2026-06-19--反應式觀察器) - [本次更新 (2026-06-19) — WCAG 2.2 稽核](#本次更新-2026-06-19--wcag-22-稽核) - [本次更新 (2026-06-19) — 記憶與決定性](#本次更新-2026-06-19--記憶與決定性) @@ -69,6 +70,13 @@ --- +## 本次更新 (2026-06-19) — SBOM 與測試分片 + +來自安全與規模研究角度的兩項純標準庫維運工具,走完整五層。完整參考:[`docs/source/Zh/doc/new_features/v18_features_doc.rst`](../docs/source/Zh/doc/new_features/v18_features_doc.rst)。 + +- **CycloneDX SBOM** — `build_sbom` / `write_sbom`(`AC_generate_sbom`、`ac_generate_sbom`):為供應鏈合規(歐盟 CRA / EO 14028)輸出 CycloneDX 1.6 相依 SBOM(name/version/purl/授權);`root` 限定某套件的封閉集,`extra_components` 可納入 action 檔。不需第三方相依。 +- **時長感知套件分片** — `shard_flows` / `merge_results`(`AC_shard_suite` / `AC_merge_results`):依每個流程歷史時長把流程裝箱成 N 片(讓最慢的 worker 而非測試數量決定總時長),再把各分片報告合併為一份彙總。 + ## 本次更新 (2026-06-19) — 反應式觀察器 非阻塞的螢幕觀察器(SikuliX `observe` 模型),走完整五層(facade、`AC_*`、MCP、Script Builder)。完整參考:[`docs/source/Zh/doc/new_features/v17_features_doc.rst`](../docs/source/Zh/doc/new_features/v17_features_doc.rst)。 diff --git a/docs/source/Eng/doc/new_features/v18_features_doc.rst b/docs/source/Eng/doc/new_features/v18_features_doc.rst new file mode 100644 index 00000000..9c86d0e2 --- /dev/null +++ b/docs/source/Eng/doc/new_features/v18_features_doc.rst @@ -0,0 +1,55 @@ +================================================== +New Features (2026-06-19) — SBOM & Suite Sharding +================================================== + +Two pure-standard-library ops tools from the security and scale research +angles, wired through the full stack (facade, ``AC_*`` executor commands, +MCP tools, Script Builder): a **CycloneDX SBOM generator** and a +**duration-aware suite sharder** (with shard-result merge). + +.. contents:: + :local: + :depth: 2 + + +CycloneDX SBOM +============= + +Supply-chain regulation (EU Cyber Resilience Act, US EO 14028) increasingly +requires a machine-readable Software Bill of Materials. ``build_sbom`` walks +the installed Python distributions and emits a **CycloneDX 1.6** JSON +document — no third-party dependency:: + + from je_auto_control import build_sbom, write_sbom + + sbom = build_sbom("je_auto_control") # dependency closure of the pkg + sbom = build_sbom(None) # every installed distribution + write_sbom("sbom.cdx.json", "je_auto_control", + extra_components=[{"type": "file", "name": "login.json", + "version": "1"}]) + +Each component carries ``name`` / ``version`` / ``purl`` (``pkg:pypi/...``) +and, when available, its license. ``extra_components`` lets you inventory +action files alongside code. Exposed as ``AC_generate_sbom`` / +``ac_generate_sbom``. + + +Duration-aware suite sharding +============================ + +Splitting a suite across N workers by *count* wastes time when tests differ +in duration — the slowest worker defines wall-clock. ``shard_flows`` balances +shards by **historical per-flow duration** from the run-history store using +greedy bin-packing:: + + from je_auto_control import shard_flows, merge_results + + shards = shard_flows(all_flows, shards=4) # ~equal time per shard + # ... each worker runs its shard, produces a report ... + report = merge_results([shard_report_1, shard_report_2, ...]) + +Flows with no history fall back to the mean of known flows (so new tests +spread evenly). ``merge_results`` recombines per-shard report dicts — summing +``total`` / ``passed`` / ``failed`` / ``skipped`` / ``errors`` and +concatenating ``results``. Exposed as ``AC_shard_suite`` / ``AC_merge_results`` +(and ``ac_shard_suite`` / ``ac_merge_results``). diff --git a/docs/source/Eng/eng_index.rst b/docs/source/Eng/eng_index.rst index 91ce935d..1dcedf1b 100644 --- a/docs/source/Eng/eng_index.rst +++ b/docs/source/Eng/eng_index.rst @@ -40,6 +40,7 @@ Comprehensive guides for all AutoControl features. doc/new_features/v15_features_doc doc/new_features/v16_features_doc doc/new_features/v17_features_doc + doc/new_features/v18_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/docs/source/Zh/doc/new_features/v18_features_doc.rst b/docs/source/Zh/doc/new_features/v18_features_doc.rst new file mode 100644 index 00000000..a84fbb86 --- /dev/null +++ b/docs/source/Zh/doc/new_features/v18_features_doc.rst @@ -0,0 +1,50 @@ +============================================== +新功能 (2026-06-19) — SBOM 與測試分片 +============================================== + +來自安全與規模研究角度的兩項純標準庫維運工具,走完整五層(facade、 +``AC_*`` 執行器指令、MCP 工具、Script Builder):**CycloneDX SBOM 產生器** +與**時長感知的測試套件分片**(含分片結果合併)。 + +.. contents:: + :local: + :depth: 2 + + +CycloneDX SBOM +============== + +供應鏈法規(歐盟網路韌性法 CRA、美國 EO 14028)日益要求可機讀的軟體物料 +清單(SBOM)。``build_sbom`` 會走訪已安裝的 Python 發行套件並輸出 +**CycloneDX 1.6** JSON 文件——不需任何第三方相依:: + + from je_auto_control import build_sbom, write_sbom + + sbom = build_sbom("je_auto_control") # 該套件的相依封閉集 + sbom = build_sbom(None) # 所有已安裝發行套件 + write_sbom("sbom.cdx.json", "je_auto_control", + extra_components=[{"type": "file", "name": "login.json", + "version": "1"}]) + +每個元件帶有 ``name`` / ``version`` / ``purl``(``pkg:pypi/...``),有提供時 +也帶授權。``extra_components`` 讓你把 action 檔與程式碼一併納入清單。對應 +``AC_generate_sbom`` / ``ac_generate_sbom``。 + + +時長感知的套件分片 +================== + +把套件依*數量*分到 N 個 worker,在測試時長不均時會浪費時間——最慢的 +worker 決定總時長。``shard_flows`` 以 run-history 中的**每個流程歷史時長** +用貪婪裝箱法平衡各分片:: + + from je_auto_control import shard_flows, merge_results + + shards = shard_flows(all_flows, shards=4) # 每片時間約略相等 + # ... 各 worker 跑自己的分片,產生一份報告 ... + report = merge_results([shard_report_1, shard_report_2, ...]) + +沒有歷史的流程退回為已知流程的平均(讓新測試平均分散)。``merge_results`` +會重新合併各分片報告 dict——加總 ``total`` / ``passed`` / ``failed`` / +``skipped`` / ``errors`` 並串接 ``results``。對應 ``AC_shard_suite`` / +``AC_merge_results``(以及 ``ac_shard_suite`` / ``ac_merge_results``)。 diff --git a/docs/source/Zh/zh_index.rst b/docs/source/Zh/zh_index.rst index 1d679bcc..08b67c5c 100644 --- a/docs/source/Zh/zh_index.rst +++ b/docs/source/Zh/zh_index.rst @@ -40,6 +40,7 @@ AutoControl 所有功能的完整使用指南。 doc/new_features/v15_features_doc doc/new_features/v16_features_doc doc/new_features/v17_features_doc + doc/new_features/v18_features_doc doc/ocr_backends/ocr_backends_doc doc/observability/observability_doc doc/operations_layer/operations_layer_doc diff --git a/je_auto_control/__init__.py b/je_auto_control/__init__.py index 9058ab34..1f2600f2 100644 --- a/je_auto_control/__init__.py +++ b/je_auto_control/__init__.py @@ -149,6 +149,10 @@ ScreenObserver, WatchRule, default_observer, image_predicate, pixel_predicate, text_predicate, ) +# CycloneDX SBOM generation (supply-chain compliance) +from je_auto_control.utils.sbom import build_sbom, write_sbom +# Duration-aware suite sharding + shard-result merge +from je_auto_control.utils.test_shard import merge_results, shard_flows # Background popup/interrupt watchdog (unattended automation) from je_auto_control.utils.watchdog import ( PopupWatchdog, WatchdogRule, default_popup_watchdog, @@ -566,6 +570,8 @@ def start_autocontrol_gui(*args, **kwargs): "DeterministicRun", "seed_everything", "ScreenObserver", "WatchRule", "default_observer", "image_predicate", "pixel_predicate", "text_predicate", + "build_sbom", "write_sbom", + "merge_results", "shard_flows", # MCP server "AuditLogger", "HttpMCPServer", "MCPContent", "MCPPrompt", "MCPPromptArgument", "MCPResource", "MCPServer", "MCPTool", diff --git a/je_auto_control/gui/script_builder/command_schema.py b/je_auto_control/gui/script_builder/command_schema.py index 589d0e44..c7553999 100644 --- a/je_auto_control/gui/script_builder/command_schema.py +++ b/je_auto_control/gui/script_builder/command_schema.py @@ -670,6 +670,33 @@ def _add_misc_specs(specs: List[CommandSpec]) -> None: description="WCAG 2.2 audit: SC-tagged findings + Target Size 2.5.8.", )) _add_observer_specs(specs) + _add_ops_specs(specs) + + +def _add_ops_specs(specs: List[CommandSpec]) -> None: + specs.append(CommandSpec( + "AC_generate_sbom", "Tools", "Generate SBOM (CycloneDX)", + fields=( + FieldSpec("path", FieldType.FILE_PATH, optional=True, + default="sbom.cdx.json"), + FieldSpec("root", FieldType.STRING, optional=True, + default="je_auto_control"), + ), + description="Emit a CycloneDX 1.6 dependency SBOM for the project.", + )) + specs.append(CommandSpec( + "AC_shard_suite", "Testing", "Shard Suite (duration-aware)", + fields=( + FieldSpec("shards", FieldType.INT, default=2), + FieldSpec("history_path", FieldType.FILE_PATH, optional=True), + FieldSpec("window", FieldType.INT, optional=True, default=20), + ), + description="Balance 'flows' (JSON view) into N shards by duration.", + )) + specs.append(CommandSpec( + "AC_merge_results", "Testing", "Merge Shard Results", + description="Merge per-shard 'reports' (JSON view) into one report.", + )) def _add_observer_specs(specs: List[CommandSpec]) -> None: diff --git a/je_auto_control/utils/executor/action_executor.py b/je_auto_control/utils/executor/action_executor.py index f0d63d02..cd532db9 100644 --- a/je_auto_control/utils/executor/action_executor.py +++ b/je_auto_control/utils/executor/action_executor.py @@ -2637,6 +2637,32 @@ def _observe_stop() -> Dict[str, Any]: return {"running": default_observer.running} +def _generate_sbom(path: Optional[str] = None, + root: str = "je_auto_control") -> Dict[str, Any]: + """Adapter: build (or write) a CycloneDX SBOM for the project.""" + from je_auto_control.utils.sbom import build_sbom, write_sbom + root_arg = root or None + if path: + return {"path": write_sbom(path, root_arg)} + return {"sbom": build_sbom(root_arg)} + + +def _shard_suite(flows: List[str], shards: int = 2, + history_path: Optional[str] = None, + window: int = 20) -> Dict[str, Any]: + """Adapter: balance flows into duration-aware shards.""" + from je_auto_control.utils.test_shard import shard_flows + return {"shards": shard_flows(flows, int(shards), + history_path=history_path, + window=int(window))} + + +def _merge_results(reports: List[Dict[str, Any]]) -> Dict[str, Any]: + """Adapter: merge per-shard report dicts into one report.""" + from je_auto_control.utils.test_shard import merge_results + return merge_results(reports) + + class Executor: """ Executor @@ -2835,6 +2861,9 @@ def __init__(self): "AC_observe_poll": _observe_poll, "AC_observe_start": _observe_start, "AC_observe_stop": _observe_stop, + "AC_generate_sbom": _generate_sbom, + "AC_shard_suite": _shard_suite, + "AC_merge_results": _merge_results, "AC_a11y_record_start": _a11y_record_start, "AC_a11y_record_stop": _a11y_record_stop, "AC_a11y_record_events": _a11y_record_events, diff --git a/je_auto_control/utils/mcp_server/tools/_factories.py b/je_auto_control/utils/mcp_server/tools/_factories.py index a4d0618b..b1ee8d88 100644 --- a/je_auto_control/utils/mcp_server/tools/_factories.py +++ b/je_auto_control/utils/mcp_server/tools/_factories.py @@ -2134,6 +2134,53 @@ def observer_tools() -> List[MCPTool]: ] +def sbom_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_generate_sbom", + description=("Generate a CycloneDX 1.6 SBOM of the project's " + "dependencies (supply-chain compliance). 'root' " + "limits to a distribution's closure (empty = all " + "installed). Writes to 'path' or returns the SBOM."), + input_schema=schema({"path": {"type": "string"}, + "root": {"type": "string"}}), + handler=h.generate_sbom, + annotations=SIDE_EFFECT_ONLY, + ), + ] + + +def sharding_tools() -> List[MCPTool]: + return [ + MCPTool( + name="ac_shard_suite", + description=("Split 'flows' into 'shards' balanced lists using " + "historical per-flow duration from run history " + "(greedy bin-pack), so each worker takes ~equal " + "time. Returns the shard lists."), + input_schema=schema({ + "flows": {"type": "array", "items": {"type": "string"}}, + "shards": {"type": "integer"}, + "history_path": {"type": "string"}, + "window": {"type": "integer"}, + }, required=["flows"]), + handler=h.shard_suite, + annotations=READ_ONLY, + ), + MCPTool( + name="ac_merge_results", + description=("Merge per-shard report dicts into one consolidated " + "report (sums total/passed/failed/skipped/errors, " + "concatenates results)."), + input_schema=schema({ + "reports": {"type": "array", "items": {"type": "object"}}, + }, required=["reports"]), + handler=h.merge_results, + annotations=READ_ONLY, + ), + ] + + def unattended_tools() -> List[MCPTool]: return [ MCPTool( @@ -3187,6 +3234,7 @@ def media_assert_tools() -> List[MCPTool]: element_repository_tools, flow_debugger_tools, skill_library_tools, guardrail_tools, a2a_tools, office_tools, agent_memory_tools, determinism_tools, observer_tools, + sbom_tools, sharding_tools, screen_record_tools, process_and_shell_tools, remote_desktop_tools, gamepad_tools, usb_passthrough_tools, assertion_tools, data_source_tools, diff --git a/je_auto_control/utils/mcp_server/tools/_handlers.py b/je_auto_control/utils/mcp_server/tools/_handlers.py index a1e69d1c..b540f3db 100644 --- a/je_auto_control/utils/mcp_server/tools/_handlers.py +++ b/je_auto_control/utils/mcp_server/tools/_handlers.py @@ -1054,6 +1054,26 @@ def observe_stop(): return {"running": default_observer.running} +def generate_sbom(path=None, root="je_auto_control"): + from je_auto_control.utils.sbom import build_sbom, write_sbom + root_arg = root or None + if path: + return {"path": write_sbom(path, root_arg)} + return {"sbom": build_sbom(root_arg)} + + +def shard_suite(flows, shards=2, history_path=None, window=20): + from je_auto_control.utils.test_shard import shard_flows + return {"shards": shard_flows(flows, int(shards), + history_path=history_path, + window=int(window))} + + +def merge_results(reports): + from je_auto_control.utils.test_shard import merge_results as _merge + return _merge(reports) + + def vlm_locate(description: str, screen_region: Optional[List[int]] = None, model: Optional[str] = None) -> Optional[List[int]]: diff --git a/je_auto_control/utils/sbom/__init__.py b/je_auto_control/utils/sbom/__init__.py new file mode 100644 index 00000000..7a2cce63 --- /dev/null +++ b/je_auto_control/utils/sbom/__init__.py @@ -0,0 +1,4 @@ +"""Software Bill of Materials (CycloneDX) generation for automation projects.""" +from je_auto_control.utils.sbom.sbom import build_sbom, write_sbom + +__all__ = ["build_sbom", "write_sbom"] diff --git a/je_auto_control/utils/sbom/sbom.py b/je_auto_control/utils/sbom/sbom.py new file mode 100644 index 00000000..3a4953e8 --- /dev/null +++ b/je_auto_control/utils/sbom/sbom.py @@ -0,0 +1,104 @@ +"""Generate a CycloneDX Software Bill of Materials for an automation project. + +Supply-chain regulation (EU Cyber Resilience Act, US EO 14028) increasingly +requires a machine-readable SBOM listing every dependency that ships with a +product. This module walks the installed Python distributions (and, +optionally, the dependency closure of one package) and emits a **CycloneDX +1.6** JSON document — the de-facto SBOM standard — without any third-party +dependency. + +Pure standard library (``importlib.metadata`` + ``json``); imports no +``PySide6``. +""" +import json +from importlib import metadata +from pathlib import Path +from typing import Any, Dict, List, Optional, Set + +_SCHEMA = "http://cyclonedx.org/schema/bom-1.6.schema.json" +_BOM_FORMAT = "CycloneDX" +_SPEC_VERSION = "1.6" + + +def _purl(name: str, version: str) -> str: + return f"pkg:pypi/{name}@{version}" + + +def _component(dist: "metadata.Distribution") -> Dict[str, Any]: + name = dist.metadata["Name"] or "unknown" + version = dist.version or "0" + component: Dict[str, Any] = { + "type": "library", "name": name, "version": version, + "purl": _purl(name, version), + } + license_name = dist.metadata.get("License") + if license_name and license_name != "UNKNOWN": + component["licenses"] = [{"license": {"name": license_name}}] + return component + + +def _iter_distributions(root: Optional[str]): + """Yield distributions: all installed, or the closure of ``root``.""" + if root is None: + yield from metadata.distributions() + return + seen: Set[str] = set() + queue = [root] + while queue: + name = queue.pop() + key = name.lower() + if key in seen: + continue + seen.add(key) + try: + dist = metadata.distribution(name) + except metadata.PackageNotFoundError: + continue + yield dist + for req in (dist.requires or []): + queue.append(_requirement_name(req)) + + +def _requirement_name(requirement: str) -> str: + """Extract the bare distribution name from a requirement string.""" + token = requirement.strip() + for sep in (" ", ";", "=", "<", ">", "!", "~", "(", "["): + token = token.split(sep, 1)[0] + return token.strip() + + +def build_sbom(root: Optional[str] = "je_auto_control", *, + extra_components: Optional[List[Dict[str, Any]]] = None + ) -> Dict[str, Any]: + """Return a CycloneDX 1.6 SBOM as a dict. + + ``root`` limits output to that distribution's dependency closure; pass + ``None`` to inventory every installed distribution. ``extra_components`` + are appended verbatim (e.g. action-file entries). + """ + components = [] + for dist in _iter_distributions(root): + try: + components.append(_component(dist)) + except (KeyError, AttributeError): + continue + components.extend(extra_components or []) + components.sort(key=lambda c: (c.get("name", ""), c.get("version", ""))) + return { + "$schema": _SCHEMA, + "bomFormat": _BOM_FORMAT, + "specVersion": _SPEC_VERSION, + "version": 1, + "metadata": {"tools": [{"name": "je_auto_control", "vendor": "JE-Chen"}]}, + "components": components, + } + + +def write_sbom(path: str = "sbom.cdx.json", + root: Optional[str] = "je_auto_control", + **kwargs: Any) -> str: + """Write a CycloneDX SBOM to ``path``; return the resolved path.""" + sbom = build_sbom(root, **kwargs) + target = Path(path) + target.write_text(json.dumps(sbom, indent=2) + "\n", encoding="utf-8") + return str(target.resolve()) diff --git a/je_auto_control/utils/test_shard/__init__.py b/je_auto_control/utils/test_shard/__init__.py new file mode 100644 index 00000000..39774a5d --- /dev/null +++ b/je_auto_control/utils/test_shard/__init__.py @@ -0,0 +1,6 @@ +"""Duration-aware suite sharding and shard-result merging.""" +from je_auto_control.utils.test_shard.test_shard import ( + merge_results, shard_flows, +) + +__all__ = ["merge_results", "shard_flows"] diff --git a/je_auto_control/utils/test_shard/test_shard.py b/je_auto_control/utils/test_shard/test_shard.py new file mode 100644 index 00000000..da62ab61 --- /dev/null +++ b/je_auto_control/utils/test_shard/test_shard.py @@ -0,0 +1,77 @@ +"""Duration-aware suite sharding (pure standard library). + +Splitting a suite across N workers by *count* wastes time when tests differ +in duration — the slowest worker defines wall-clock. This balances shards by +**historical per-flow duration** (from the run-history store) using greedy +bin-packing, so each shard takes roughly the same time. :func:`merge_results` +recombines the per-shard reports afterwards — the standard companion step. + +Imports no ``PySide6``. +""" +from typing import Any, Dict, List, Optional + +_SUM_KEYS = ("total", "passed", "failed", "skipped", "errors") + + +def _durations(flows: List[str], history_path: Optional[str], + window: int) -> Dict[str, float]: + """Mean recent wall-clock seconds per flow, from run history.""" + from je_auto_control.utils.run_history import ( + HistoryStore, default_history_store) + store, owned = ((HistoryStore(history_path), True) if history_path + else (default_history_store, False)) + try: + records = store.list_runs( + limit=max(100, int(window) * max(1, len(flows)))) + finally: + if owned: + store.close() + samples: Dict[str, List[float]] = {} + for record in records: + seconds = record.duration_seconds + if seconds is not None: + samples.setdefault(record.script_path, []).append(seconds) + return {flow: sum(values[:window]) / len(values[:window]) + for flow, values in samples.items() if values} + + +def shard_flows(flows: List[str], shards: int, *, + history_path: Optional[str] = None, window: int = 20, + default_weight: Optional[float] = None) -> List[List[str]]: + """Split ``flows`` into ``shards`` lists balanced by mean duration. + + Flows with no history use ``default_weight`` (or the mean of known flows, + else 1.0). Greedy: heaviest flow first onto the currently-lightest shard. + """ + flows = list(flows) + count = max(1, int(shards)) + means = _durations(flows, history_path, int(window)) + known = list(means.values()) + fallback = (default_weight if default_weight is not None + else (sum(known) / len(known) if known else 1.0)) + ordered = sorted(flows, key=lambda f: means.get(f, fallback), reverse=True) + buckets: List[List[str]] = [[] for _ in range(count)] + loads = [0.0] * count + for flow in ordered: + index = min(range(count), key=lambda k: loads[k]) + buckets[index].append(flow) + loads[index] += means.get(flow, fallback) + return buckets + + +def merge_results(reports: List[Dict[str, Any]]) -> Dict[str, Any]: + """Merge per-shard report dicts into one consolidated report. + + Numeric keys (total/passed/failed/skipped/errors) are summed and + ``results`` lists concatenated. + """ + reports = list(reports) + merged: Dict[str, Any] = {key: 0 for key in _SUM_KEYS} + results: List[Any] = [] + for report in reports: + for key in _SUM_KEYS: + merged[key] += int(report.get(key, 0) or 0) + results.extend(report.get("results", []) or []) + merged["shards"] = len(reports) + merged["results"] = results + return merged diff --git a/test/unit_test/headless/test_ops_batch.py b/test/unit_test/headless/test_ops_batch.py new file mode 100644 index 00000000..9dabb5a8 --- /dev/null +++ b/test/unit_test/headless/test_ops_batch.py @@ -0,0 +1,93 @@ +"""Headless tests for the ops batch: CycloneDX SBOM generation and +duration-aware suite sharding + shard-result merge. Pure stdlib; no Qt.""" +import json + +import je_auto_control as ac +from je_auto_control.utils.sbom import build_sbom, write_sbom +from je_auto_control.utils.test_shard import merge_results, shard_flows + + +# --- SBOM ----------------------------------------------------------------- + +def test_sbom_core_shape(): + sbom = build_sbom("je_auto_control") + assert sbom["bomFormat"] == "CycloneDX" + assert sbom["specVersion"] == "1.6" + assert isinstance(sbom["components"], list) and sbom["components"] + comp = sbom["components"][0] + assert {"type", "name", "version", "purl"} <= set(comp) + assert comp["purl"].startswith("pkg:pypi/") + + +def test_sbom_extra_components_and_write(tmp_path): + extra = [{"type": "file", "name": "login.json", "version": "1"}] + sbom = build_sbom("je_auto_control", extra_components=extra) + assert any(c["name"] == "login.json" for c in sbom["components"]) + path = write_sbom(str(tmp_path / "s.cdx.json"), "je_auto_control") + assert json.loads(open(path, encoding="utf-8").read())["specVersion"] == \ + "1.6" + + +# --- suite sharding ------------------------------------------------------- + +def test_shard_balances_by_duration(tmp_path): + from je_auto_control.utils.run_history import HistoryStore + db = str(tmp_path / "h.sqlite") + store = HistoryStore(db) + # "slow" ~ long duration, three "fast" ~ short. + durations = {"slow": 9.0, "f1": 1.0, "f2": 1.0, "f3": 1.0} + for flow, secs in durations.items(): + rid = store.start_run("manual", "x", flow, started_at=1000.0) + store.finish_run(rid, "ok", finished_at=1000.0 + secs) + store.close() + shards = shard_flows(list(durations), 2, history_path=db) + assert len(shards) == 2 + # the heavy flow is alone; the three light flows share the other shard + heavy = [s for s in shards if "slow" in s][0] + assert heavy == ["slow"] + assert sorted(s for s in shards if s != heavy)[0] == ["f1", "f2", "f3"] + + +def test_shard_unknown_flows_spread_evenly(): + shards = shard_flows(["a", "b", "c", "d"], 2) # no history + assert len(shards) == 2 + assert sorted(len(s) for s in shards) == [2, 2] + + +def test_merge_results_sums_and_concatenates(): + merged = merge_results([ + {"total": 3, "passed": 2, "failed": 1, "results": ["a", "b"]}, + {"total": 2, "passed": 2, "failed": 0, "results": ["c"]}, + ]) + assert merged["total"] == 5 and merged["passed"] == 4 + assert merged["failed"] == 1 and merged["shards"] == 2 + assert merged["results"] == ["a", "b", "c"] + + +# --- wiring --------------------------------------------------------------- + +def test_executor_wiring(tmp_path): + rec = ac.execute_action([["AC_generate_sbom", { + "path": str(tmp_path / "e.cdx.json"), "root": "je_auto_control"}]]) + assert any("path" in str(v) for v in rec.values()) + sh = ac.execute_action([["AC_shard_suite", { + "flows": ["a", "b", "c"], "shards": 2}]]) + assert any("shards" in str(v) for v in sh.values()) + known = ac.executor.known_commands() + assert {"AC_generate_sbom", "AC_shard_suite", "AC_merge_results"} <= known + + +def test_mcp_and_builder_wiring(): + from je_auto_control.utils.mcp_server.tools import ( + build_default_tool_registry) + names = {t.name for t in build_default_tool_registry()} + assert {"ac_generate_sbom", "ac_shard_suite", "ac_merge_results"} <= names + from je_auto_control.gui.script_builder.command_schema import _build_specs + cmds = {s.command for s in _build_specs()} + assert {"AC_generate_sbom", "AC_shard_suite", "AC_merge_results"} <= cmds + + +def test_facade_exports(): + for attr in ("build_sbom", "write_sbom", "shard_flows", "merge_results"): + assert hasattr(ac, attr) + assert attr in ac.__all__ From 68fdb98d8594f8fc15538ec601f290008894247b Mon Sep 17 00:00:00 2001 From: JeffreyChen Date: Fri, 19 Jun 2026 11:45:59 +0800 Subject: [PATCH 2/2] Use https schema URL and clear Sonar code smells in ops batch --- je_auto_control/utils/sbom/sbom.py | 2 +- je_auto_control/utils/test_shard/test_shard.py | 10 +++++++--- test/unit_test/headless/test_ops_batch.py | 3 ++- 3 files changed, 10 insertions(+), 5 deletions(-) diff --git a/je_auto_control/utils/sbom/sbom.py b/je_auto_control/utils/sbom/sbom.py index 3a4953e8..5f3ae282 100644 --- a/je_auto_control/utils/sbom/sbom.py +++ b/je_auto_control/utils/sbom/sbom.py @@ -15,7 +15,7 @@ from pathlib import Path from typing import Any, Dict, List, Optional, Set -_SCHEMA = "http://cyclonedx.org/schema/bom-1.6.schema.json" +_SCHEMA = "https://cyclonedx.org/schema/bom-1.6.schema.json" _BOM_FORMAT = "CycloneDX" _SPEC_VERSION = "1.6" diff --git a/je_auto_control/utils/test_shard/test_shard.py b/je_auto_control/utils/test_shard/test_shard.py index da62ab61..ef0f4b11 100644 --- a/je_auto_control/utils/test_shard/test_shard.py +++ b/je_auto_control/utils/test_shard/test_shard.py @@ -47,8 +47,12 @@ def shard_flows(flows: List[str], shards: int, *, count = max(1, int(shards)) means = _durations(flows, history_path, int(window)) known = list(means.values()) - fallback = (default_weight if default_weight is not None - else (sum(known) / len(known) if known else 1.0)) + if default_weight is not None: + fallback = float(default_weight) + elif known: + fallback = sum(known) / len(known) + else: + fallback = 1.0 ordered = sorted(flows, key=lambda f: means.get(f, fallback), reverse=True) buckets: List[List[str]] = [[] for _ in range(count)] loads = [0.0] * count @@ -66,7 +70,7 @@ def merge_results(reports: List[Dict[str, Any]]) -> Dict[str, Any]: ``results`` lists concatenated. """ reports = list(reports) - merged: Dict[str, Any] = {key: 0 for key in _SUM_KEYS} + merged: Dict[str, Any] = dict.fromkeys(_SUM_KEYS, 0) results: List[Any] = [] for report in reports: for key in _SUM_KEYS: diff --git a/test/unit_test/headless/test_ops_batch.py b/test/unit_test/headless/test_ops_batch.py index 9dabb5a8..30790692 100644 --- a/test/unit_test/headless/test_ops_batch.py +++ b/test/unit_test/headless/test_ops_batch.py @@ -45,7 +45,8 @@ def test_shard_balances_by_duration(tmp_path): # the heavy flow is alone; the three light flows share the other shard heavy = [s for s in shards if "slow" in s][0] assert heavy == ["slow"] - assert sorted(s for s in shards if s != heavy)[0] == ["f1", "f2", "f3"] + other = [s for s in shards if s != heavy][0] + assert sorted(other) == ["f1", "f2", "f3"] def test_shard_unknown_flows_spread_evenly():