run plugin validation concurrently

Author: zouyonghe

scripts/validate_plugins/run.py

@@ -4,6 +4,7 @@
 
 import argparse
 import asyncio
+import concurrent.futures
 import hashlib
 import json
 import os
@@ -29,6 +30,7 @@
 
 REQUIRED_METADATA_FIELDS = ("name", "desc", "version", "author")
 DEFAULT_CLONE_TIMEOUT = 120
+DEFAULT_MAX_WORKERS = 8
 CONFLICT_MARKERS = ("<<<<<<<", "=======", ">>>>>>>")
 
 
@@ -489,29 +491,60 @@ def validate_selected_plugins(
     work_dir: Path,
     clone_timeout: int,
     load_timeout: int,
+    max_workers: int,
 ) -> list[dict]:
-    results = []
     total = len(selected)
-
-    for index, (plugin, plugin_data) in enumerate(selected, start=1):
-        print(f"[{index}/{total}] Validating {plugin}", flush=True)
-        result = validate_plugin(
-            plugin=plugin,
-            plugin_data=plugin_data,
-            astrbot_path=astrbot_path,
-            script_path=script_path,
-            work_dir=work_dir,
-            clone_timeout=clone_timeout,
-            load_timeout=load_timeout,
+    results: list[dict | None] = [None] * total
+
+    def task(index: int, plugin: str, plugin_data: dict) -> tuple[int, dict]:
+        return (
+            index,
+            validate_plugin(
+                plugin=plugin,
+                plugin_data=plugin_data,
+                astrbot_path=astrbot_path,
+                script_path=script_path,
+                work_dir=work_dir,
+                clone_timeout=clone_timeout,
+                load_timeout=load_timeout,
+            ),
         )
-        results.append(result)
 
-        status = "PASS" if result.get("ok") else "FAIL"
-        stage = result.get("stage", "unknown")
-        message = result.get("message", "")
-        print(f"[{index}/{total}] {status} {plugin} [{stage}] {message}", flush=True)
+    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
+        future_to_context: dict[concurrent.futures.Future, tuple[int, str]] = {}
 
-    return results
+        for index, (plugin, plugin_data) in enumerate(selected, start=1):
+            print(f"[{index}/{total}] Queued {plugin}", flush=True)
+            future = executor.submit(task, index, plugin, plugin_data)
+            future_to_context[future] = (index, plugin)
+
+        for future in concurrent.futures.as_completed(future_to_context):
+            index, plugin = future_to_context[future]
+            try:
+                original_index, result = future.result()
+            except Exception as exc:
+                original_index = index
+                result = build_result(
+                    plugin=plugin,
+                    repo="",
+                    normalized_repo_url=None,
+                    ok=False,
+                    stage="threadpool",
+                    message=str(exc),
+                    details=traceback.format_exc(),
+                )
+
+            results[original_index - 1] = result
+            status = "PASS" if result.get("ok") else "FAIL"
+            stage = result.get("stage", "unknown")
+            message = result.get("message", "")
+            print(f"[{original_index}/{total}] {status} {plugin} [{stage}] {message}", flush=True)
+
+    finalized = [result for result in results if result is not None]
+    if len(finalized) != total:
+        raise RuntimeError("parallel validation finished with missing results")
+
+    return finalized
 
 
 class NullStub:
@@ -684,6 +717,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--work-dir")
     parser.add_argument("--clone-timeout", type=int, default=DEFAULT_CLONE_TIMEOUT)
     parser.add_argument("--load-timeout", type=int, default=300)
+    parser.add_argument("--max-workers", type=int, default=DEFAULT_MAX_WORKERS)
     parser.add_argument("--worker", action="store_true")
     parser.add_argument("--plugin-source-dir")
     parser.add_argument("--plugin-dir-name")
@@ -736,6 +770,7 @@ def main() -> int:
             work_dir=work_dir,
             clone_timeout=args.clone_timeout,
             load_timeout=args.load_timeout,
+            max_workers=args.max_workers,
         )
     finally:
         if temp_dir is not None:

tests/test_validate_plugins.py

@@ -386,6 +386,13 @@ def test_load_plugins_index_rejects_non_dict_values(self):
 
 
 class ValidationProgressTests(unittest.TestCase):
+    def test_build_parser_defaults_max_workers_to_eight(self):
+        module = load_validator_module()
+
+        args = module.build_parser().parse_args(["--astrbot-path", "/tmp/AstrBot"])
+
+        self.assertEqual(args.max_workers, 8)
+
     def test_validate_selected_plugins_emits_progress_and_result_lines(self):
         module = load_validator_module()
         selected = [
@@ -406,14 +413,58 @@ def test_validate_selected_plugins_emits_progress_and_result_lines(self):
                     work_dir=Path("/tmp/work"),
                     clone_timeout=60,
                     load_timeout=300,
+                    max_workers=8,
                 )
 
         self.assertEqual(results, fake_results)
         self.assertEqual(validate_mock.call_count, 2)
-        print_mock.assert_any_call("[1/2] Validating plugin-a", flush=True)
+        print_mock.assert_any_call("[1/2] Queued plugin-a", flush=True)
         print_mock.assert_any_call("[1/2] PASS plugin-a [load] ok", flush=True)
         print_mock.assert_any_call("[2/2] FAIL plugin-b [metadata] invalid metadata.yaml", flush=True)
 
+    def test_validate_selected_plugins_preserves_result_order_with_out_of_order_completion(self):
+        module = load_validator_module()
+        selected = [
+            ("plugin-a", {"repo": "https://github.com/example/plugin-a"}),
+            ("plugin-b", {"repo": "https://github.com/example/plugin-b"}),
+            ("plugin-c", {"repo": "https://github.com/example/plugin-c"}),
+        ]
+        futures = [mock.Mock(name="future-a"), mock.Mock(name="future-b"), mock.Mock(name="future-c")]
+        future_to_result = {
+            futures[0]: (1, {"plugin": "plugin-a", "ok": True, "stage": "load", "message": "a"}),
+            futures[1]: (2, {"plugin": "plugin-b", "ok": False, "stage": "metadata", "message": "b"}),
+            futures[2]: (3, {"plugin": "plugin-c", "ok": True, "stage": "load", "message": "c"}),
+        }
+
+        executor = mock.MagicMock()
+        executor.__enter__.return_value = executor
+        executor.__exit__.return_value = False
+        executor.submit.side_effect = futures
+
+        def future_result(future):
+            return future_to_result[future]
+
+        for future in futures:
+            future.result.side_effect = lambda _timeout=None, future=future: future_result(future)
+
+        with mock.patch.object(module.concurrent.futures, "ThreadPoolExecutor", return_value=executor) as pool_mock:
+            with mock.patch.object(module.concurrent.futures, "as_completed", return_value=[futures[2], futures[0], futures[1]]):
+                with mock.patch("builtins.print") as print_mock:
+                    results = module.validate_selected_plugins(
+                        selected=selected,
+                        astrbot_path=Path("/tmp/AstrBot"),
+                        script_path=Path("/tmp/run.py"),
+                        work_dir=Path("/tmp/work"),
+                        clone_timeout=60,
+                        load_timeout=300,
+                        max_workers=8,
+                    )
+
+        pool_mock.assert_called_once_with(max_workers=8)
+        self.assertEqual([item["plugin"] for item in results], ["plugin-a", "plugin-b", "plugin-c"])
+        print_mock.assert_any_call("[1/3] Queued plugin-a", flush=True)
+        print_mock.assert_any_call("[3/3] PASS plugin-c [load] c", flush=True)
+
 
 class ValidatePluginTests(unittest.TestCase):
     def setUp(self):