tighten validator CLI and warning semantics

Author: zouyonghe

scripts/validate_plugins/run.py

@@ -38,6 +38,13 @@ class MetadataLoadError(ValueError):
     pass
 
 
+def positive_int(raw_value: str) -> int:
+    value = int(raw_value)
+    if value <= 0:
+        raise argparse.ArgumentTypeError("must be a positive integer")
+    return value
+
+
 def build_result(
     *,
     plugin: str,
@@ -50,14 +57,16 @@ def build_result(
     plugin_dir_name: str | None = None,
     details: dict | str | None = None,
 ) -> dict:
+    resolved_severity = severity or ("pass" if ok else "fail")
+    resolved_ok = True if resolved_severity in {"pass", "warn"} else ok
     result = {
         "plugin": plugin,
         "repo": repo,
         "normalized_repo_url": normalized_repo_url,
-        "ok": ok,
+        "ok": resolved_ok,
         "stage": stage,
         "message": message,
-        "severity": severity or ("pass" if ok else "fail"),
+        "severity": resolved_severity,
     }
     if plugin_dir_name:
         result["plugin_dir_name"] = plugin_dir_name
@@ -742,9 +751,9 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--astrbot-path")
     parser.add_argument("--report-path", default="validation-report.json")
     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("--clone-timeout", type=positive_int, default=DEFAULT_CLONE_TIMEOUT)
+    parser.add_argument("--load-timeout", type=positive_int, default=300)
+    parser.add_argument("--max-workers", type=positive_int, default=DEFAULT_MAX_WORKERS)
     parser.add_argument("--worker", action="store_true")
     parser.add_argument("--plugin-source-dir")
     parser.add_argument("--plugin-dir-name")

tests/test_validate_plugins.py

@@ -393,6 +393,18 @@ def test_build_parser_defaults_max_workers_to_eight(self):
 
         self.assertEqual(args.max_workers, 8)
 
+    def test_build_parser_rejects_non_positive_worker_and_timeout_values(self):
+        module = load_validator_module()
+
+        with self.assertRaises(SystemExit):
+            module.build_parser().parse_args(["--astrbot-path", "/tmp/AstrBot", "--max-workers", "0"])
+
+        with self.assertRaises(SystemExit):
+            module.build_parser().parse_args(["--astrbot-path", "/tmp/AstrBot", "--clone-timeout", "0"])
+
+        with self.assertRaises(SystemExit):
+            module.build_parser().parse_args(["--astrbot-path", "/tmp/AstrBot", "--load-timeout", "0"])
+
     def test_validate_selected_plugins_emits_progress_and_result_lines(self):
         module = load_validator_module()
         selected = [
@@ -546,6 +558,51 @@ def test_precheck_failure_is_mapped_into_result(self):
         self.assertEqual(result["message"], "invalid metadata")
         self.assertEqual(result["details"], "line 3")
 
+    def test_precheck_warning_is_non_fatal_in_final_result(self):
+        with mock.patch.object(self.module, "clone_plugin_repo"):
+            with mock.patch.object(
+                self.module,
+                "precheck_plugin_directory",
+                return_value={
+                    "ok": False,
+                    "severity": "warn",
+                    "stage": "metadata",
+                    "message": "missing required metadata fields: desc",
+                },
+            ):
+                result = self.call_validate_plugin()
+
+        self.assertTrue(result["ok"])
+        self.assertEqual(result["severity"], "warn")
+        self.assertEqual(result["stage"], "metadata")
+
+    def test_load_timeout_uses_process_output_details(self):
+        timeout = subprocess.TimeoutExpired(
+            cmd=[sys.executable, str(self.script_path)],
+            timeout=60,
+            output="timeout-stdout",
+            stderr="timeout-stderr",
+        )
+
+        with mock.patch.object(
+            self.module,
+            "precheck_plugin_directory",
+            return_value={"ok": True, "plugin_dir_name": "demo-plugin", "message": "ok", "stage": "precheck"},
+        ):
+            with mock.patch.object(self.module, "clone_plugin_repo"):
+                with mock.patch.object(subprocess, "run", side_effect=timeout):
+                    with mock.patch.object(
+                        self.module,
+                        "build_process_output_details",
+                        return_value={"stdout": "timeout-stdout", "stderr": "timeout-stderr"},
+                    ) as details_mock:
+                        result = self.call_validate_plugin()
+
+        self.assertEqual(result["stage"], "timeout")
+        self.assertEqual(result["plugin_dir_name"], "demo-plugin")
+        self.assertEqual(result["details"], {"stdout": "timeout-stdout", "stderr": "timeout-stderr"})
+        details_mock.assert_called_once_with(stdout="timeout-stdout", stderr="timeout-stderr")
+
     def test_successful_clone_and_precheck_invokes_worker_and_parses_output(self):
         completed = subprocess.CompletedProcess(
             args=["python3", "run.py"],