improve plugin validation diagnostics

Author: zouyonghe

.github/workflows/validate-plugin-smoke.yml

@@ -46,6 +46,7 @@ jobs:
           import json
           import os
           import subprocess
+          import sys
           from pathlib import Path
 
           base_ref = os.environ["GITHUB_BASE_REF"]
@@ -70,9 +71,11 @@ jobs:
               if not isinstance(head, dict):
                   raise ValueError("plugins.json must contain a JSON object")
           except (json.JSONDecodeError, ValueError) as exc:
-              changed = []
-              should_validate = False
-              validation_note = f"Skipping smoke validation because plugins.json is invalid on the PR head: {exc}"
+              print(
+                  f"plugins.json is invalid on the PR head; smoke validation cannot continue: {exc}",
+                  file=sys.stderr,
+              )
+              raise SystemExit(1)
           else:
               changed = [
                   name

scripts/validate_plugins/run.py

@@ -98,13 +98,27 @@ def select_plugins(
 
 
 def _parse_simple_yaml(path: Path) -> dict:
+    def parse_value(raw_value: str) -> str:
+        value = raw_value.strip()
+        if not value:
+            return ""
+
+        if value[0] in {'"', "'"}:
+            quote = value[0]
+            end_index = value.rfind(quote)
+            if end_index > 0:
+                return value[1:end_index]
+
+        value = re.split(r"\s+#", value, maxsplit=1)[0].rstrip()
+        return value.strip("\"'")
+
     result = {}
     for raw_line in path.read_text(encoding="utf-8").splitlines():
         line = raw_line.strip()
         if not line or line.startswith("#") or ":" not in line:
             continue
         key, value = line.split(":", 1)
-        result[key.strip()] = value.strip().strip("\"'")
+        result[key.strip()] = parse_value(value)
     return result
 
 
@@ -219,11 +233,17 @@ def load_plugins_index(path: Path) -> dict[str, dict]:
     data = json.loads(path.read_text(encoding="utf-8"))
     if not isinstance(data, dict):
         raise ValueError("plugins.json must contain a JSON object")
-    result = {}
+
     for key, value in data.items():
-        if isinstance(key, str) and isinstance(value, dict):
-            result[key] = value
-    return result
+        if not isinstance(key, str):
+            raise ValueError(
+                f"plugins.json keys must be strings, got {type(key).__name__!r}: {key!r}"
+            )
+        if not isinstance(value, dict):
+            raise ValueError(
+                f"plugins.json values must be objects/dicts; key {key!r} has {type(value).__name__!r}"
+            )
+    return data
 
 
 def combine_requested_names(
@@ -431,6 +451,39 @@ def validate_plugin(
     )
 
 
+def validate_selected_plugins(
+    *,
+    selected: list[tuple[str, dict]],
+    astrbot_path: Path,
+    script_path: Path,
+    work_dir: Path,
+    clone_timeout: int,
+    load_timeout: 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.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)
+
+    return results
+
+
 class NullStub:
     def __getattr__(self, name: str) -> "NullStub":
         del name
@@ -646,18 +699,14 @@ def main() -> int:
     work_dir.mkdir(parents=True, exist_ok=True)
 
     try:
-        results = [
-            validate_plugin(
-                plugin=plugin,
-                plugin_data=plugin_data,
-                astrbot_path=Path(args.astrbot_path).resolve(),
-                script_path=Path(__file__).resolve(),
-                work_dir=work_dir,
-                clone_timeout=args.clone_timeout,
-                load_timeout=args.load_timeout,
-            )
-            for plugin, plugin_data in selected
-        ]
+        results = validate_selected_plugins(
+            selected=selected,
+            astrbot_path=Path(args.astrbot_path).resolve(),
+            script_path=Path(__file__).resolve(),
+            work_dir=work_dir,
+            clone_timeout=args.clone_timeout,
+            load_timeout=args.load_timeout,
+        )
     finally:
         if temp_dir is not None:
             temp_dir.cleanup()

tests/test_validate_plugins.py

@@ -1,9 +1,12 @@
 import importlib.util
+import json
+import os
 import subprocess
 import sys
 import tempfile
 import unittest
 from pathlib import Path
+from unittest import mock
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -195,6 +198,154 @@ def test_build_process_output_details_keeps_partial_timeout_logs(self):
 
         self.assertEqual(details, {"stdout": "line one\nline two", "stderr": "warning"})
 
+    def test_parse_simple_yaml_handles_comments_quotes_and_whitespace(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write(
+                """
+                # leading comment
+
+                key1: value1      # trailing comment
+                key2: \" spaced value \"
+                key3: 'another value'
+                key4: value-with-#-hash
+                """
+            )
+            metadata_path = Path(handle.name)
+
+        try:
+            parsed = module._parse_simple_yaml(metadata_path)
+        finally:
+            os.remove(metadata_path)
+
+        self.assertEqual(parsed["key1"], "value1")
+        self.assertEqual(parsed["key2"], " spaced value ")
+        self.assertEqual(parsed["key3"], "another value")
+        self.assertEqual(parsed["key4"], "value-with-#-hash")
+
+    def test_load_metadata_uses_yaml_safe_load_when_available(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write("name: should-be-overridden\n")
+            metadata_path = Path(handle.name)
+
+        fake_yaml = mock.Mock()
+        fake_yaml.safe_load.return_value = {"name": "from-yaml", "version": "1.0.0"}
+
+        try:
+            with mock.patch.object(module, "yaml", fake_yaml):
+                metadata = module.load_metadata(metadata_path)
+        finally:
+            os.remove(metadata_path)
+
+        self.assertEqual(metadata, {"name": "from-yaml", "version": "1.0.0"})
+        fake_yaml.safe_load.assert_called_once()
+
+    def test_load_metadata_uses_simple_parser_when_yaml_unavailable(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".yml", delete=False) as handle:
+            handle.write(
+                """
+                name: demo-plugin
+                version: \"0.2.3\"
+                """
+            )
+            metadata_path = Path(handle.name)
+
+        yaml_backup = getattr(module, "yaml", None)
+        try:
+            module.yaml = None
+            metadata = module.load_metadata(metadata_path)
+        finally:
+            module.yaml = yaml_backup
+            os.remove(metadata_path)
+
+        self.assertEqual(metadata.get("name"), "demo-plugin")
+        self.assertEqual(metadata.get("version"), "0.2.3")
+
+    def test_load_plugins_index_accepts_valid_object(self):
+        module = load_validator_module()
+
+        index_obj = {
+            "good-plugin": {"name": "Good Plugin", "repo": "https://github.com/example/good"},
+            "another-plugin": {"name": "Another Plugin"},
+        }
+
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
+            json.dump(index_obj, handle)
+            index_path = Path(handle.name)
+
+        try:
+            plugins = module.load_plugins_index(index_path)
+        finally:
+            os.remove(index_path)
+
+        self.assertEqual(plugins, index_obj)
+
+    def test_load_plugins_index_rejects_json_array(self):
+        module = load_validator_module()
+
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
+            json.dump([{"name": "array-entry"}], handle)
+            index_path = Path(handle.name)
+
+        try:
+            with self.assertRaisesRegex(ValueError, "plugins.json must contain a JSON object"):
+                module.load_plugins_index(index_path)
+        finally:
+            os.remove(index_path)
+
+    def test_load_plugins_index_rejects_non_dict_values(self):
+        module = load_validator_module()
+
+        index_obj = {
+            "valid-plugin": {"name": "Valid Plugin", "repo": "https://github.com/example/valid"},
+            "not-a-dict": "just-a-string",
+        }
+
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
+            json.dump(index_obj, handle)
+            index_path = Path(handle.name)
+
+        try:
+            with self.assertRaisesRegex(ValueError, "plugins.json values must be objects/dicts"):
+                module.load_plugins_index(index_path)
+        finally:
+            os.remove(index_path)
+
+
+class ValidationProgressTests(unittest.TestCase):
+    def test_validate_selected_plugins_emits_progress_and_result_lines(self):
+        module = load_validator_module()
+        selected = [
+            ("plugin-a", {"repo": "https://github.com/example/plugin-a"}),
+            ("plugin-b", {"repo": "https://github.com/example/plugin-b"}),
+        ]
+        fake_results = [
+            {"plugin": "plugin-a", "ok": True, "stage": "load", "message": "ok"},
+            {"plugin": "plugin-b", "ok": False, "stage": "metadata", "message": "invalid metadata.yaml"},
+        ]
+
+        with mock.patch.object(module, "validate_plugin", side_effect=fake_results) as validate_mock:
+            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,
+                )
+
+        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] PASS plugin-a [load] ok", flush=True)
+        print_mock.assert_any_call("[2/2] FAIL plugin-b [metadata] invalid metadata.yaml", flush=True)
+
 
 class MetadataValidationTests(unittest.TestCase):
     def test_reports_missing_required_metadata_fields(self):