harden plugin smoke validation edge cases

Author: zouyonghe

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

@@ -50,32 +50,52 @@ jobs:
 
           base_ref = os.environ["GITHUB_BASE_REF"]
           subprocess.run(["git", "fetch", "origin", base_ref, "--depth", "1"], check=True)
-          base = json.loads(
-              subprocess.check_output(
+
+          validation_note = ""
+
+          try:
+              base_raw = subprocess.check_output(
                   ["git", "show", f"origin/{base_ref}:plugins.json"],
                   text=True,
+                  stderr=subprocess.DEVNULL,
               )
-          )
-          head = json.loads(Path("plugins.json").read_text(encoding="utf-8"))
-
-          changed = [
-              name
-              for name, payload in head.items()
-              if base.get(name) != payload
-          ]
+              base = json.loads(base_raw)
+              if not isinstance(base, dict):
+                  base = {}
+          except (subprocess.CalledProcessError, json.JSONDecodeError):
+              base = {}
+
+          try:
+              head = json.loads(Path("plugins.json").read_text(encoding="utf-8"))
+              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}"
+          else:
+              changed = [
+                  name
+                  for name, payload in head.items()
+                  if base.get(name) != payload
+              ]
+              should_validate = bool(changed)
+              if not should_validate:
+                  validation_note = "No plugin entries changed in plugins.json; skipping smoke validation."
 
           with open(os.environ["GITHUB_ENV"], "a", encoding="utf-8") as handle:
               handle.write("ASTRBOT_REF=master\n")
               handle.write(f"PLUGIN_NAME_LIST={','.join(changed)}\n")
               handle.write("PLUGIN_LIMIT=\n")
-              handle.write(f"SHOULD_VALIDATE={'true' if changed else 'false'}\n")
+              handle.write(f"SHOULD_VALIDATE={'true' if should_validate else 'false'}\n")
+              handle.write(f"VALIDATION_NOTE={validation_note}\n")
           PY
 
       - name: Show PR diff selection
         if: github.event_name == 'pull_request'
         run: |
           if [ "$SHOULD_VALIDATE" != "true" ]; then
-            printf '%s\n' "No plugin entries changed in plugins.json; skipping smoke validation."
+            printf '%s\n' "${VALIDATION_NOTE:-Smoke validation skipped.}"
           else
             printf 'Selected plugins: %s\n' "$PLUGIN_NAME_LIST"
           fi

scripts/validate_plugins/run.py

@@ -4,6 +4,7 @@
 
 import argparse
 import asyncio
+import hashlib
 import json
 import os
 import re
@@ -22,6 +23,7 @@
 
 
 REQUIRED_METADATA_FIELDS = ("name", "desc", "version", "author")
+DEFAULT_CLONE_TIMEOUT = 120
 
 
 def build_result(
@@ -205,7 +207,7 @@ def combine_requested_names(
     plugin_names: list[str] | None,
     plugin_name_list: str | None,
 ) -> list[str]:
-    names = list(plugin_names or [])
+    names = [name.strip() for name in (plugin_names or [])]
     if plugin_name_list:
         names.extend(part.strip() for part in plugin_name_list.split(","))
     return [name for name in names if name]
@@ -216,12 +218,47 @@ def sanitize_name(name: str) -> str:
     return sanitized or "plugin"
 
 
-def clone_plugin_repo(repo_url: str, destination: Path) -> None:
+def build_plugin_clone_dir(work_dir: Path, plugin: str) -> Path:
+    digest = hashlib.sha256(plugin.encode("utf-8")).hexdigest()[:8]
+    return work_dir / f"{sanitize_name(plugin)}-{digest}"
+
+
+def _normalize_process_output(output: str | bytes | None) -> str | None:
+    if output is None:
+        return None
+    if isinstance(output, bytes):
+        output = output.decode("utf-8", errors="replace")
+    normalized = output.strip()
+    return normalized or None
+
+
+def build_process_output_details(
+    *,
+    stdout: str | bytes | None,
+    stderr: str | bytes | None,
+) -> dict | None:
+    details = {}
+    stdout_text = _normalize_process_output(stdout)
+    stderr_text = _normalize_process_output(stderr)
+    if stdout_text:
+        details["stdout"] = stdout_text
+    if stderr_text:
+        details["stderr"] = stderr_text
+    return details or None
+
+
+def clone_plugin_repo(
+    repo_url: str,
+    destination: Path,
+    *,
+    timeout: int = DEFAULT_CLONE_TIMEOUT,
+) -> None:
     subprocess.run(
         ["git", "clone", "--depth", "1", repo_url, str(destination)],
         check=True,
         capture_output=True,
         text=True,
+        timeout=timeout,
     )
 
 
@@ -267,6 +304,7 @@ def validate_plugin(
     astrbot_path: Path,
     script_path: Path,
     work_dir: Path,
+    clone_timeout: int,
     load_timeout: int,
 ) -> dict:
     repo_url = plugin_data.get("repo")
@@ -292,9 +330,13 @@ def validate_plugin(
             message=str(exc),
         )
 
-    plugin_clone_dir = work_dir / sanitize_name(plugin)
+    plugin_clone_dir = build_plugin_clone_dir(work_dir, plugin)
     try:
-        clone_plugin_repo(normalized_repo_url, plugin_clone_dir)
+        clone_plugin_repo(
+            normalized_repo_url,
+            plugin_clone_dir,
+            timeout=clone_timeout,
+        )
     except subprocess.CalledProcessError as exc:
         message = exc.stderr.strip() or exc.stdout.strip() or str(exc)
         return build_result(
@@ -305,6 +347,16 @@ def validate_plugin(
             stage="clone",
             message=message,
         )
+    except subprocess.TimeoutExpired as exc:
+        return build_result(
+            plugin=plugin,
+            repo=repo_url,
+            normalized_repo_url=normalized_repo_url,
+            ok=False,
+            stage="clone_timeout",
+            message=f"git clone timed out after {clone_timeout} seconds",
+            details=build_process_output_details(stdout=exc.stdout, stderr=exc.stderr),
+        )
 
     precheck = precheck_plugin_directory(plugin_clone_dir)
     if not precheck["ok"]:
@@ -334,7 +386,7 @@ def validate_plugin(
             text=True,
             timeout=load_timeout,
         )
-    except subprocess.TimeoutExpired:
+    except subprocess.TimeoutExpired as exc:
         return build_result(
             plugin=plugin,
             repo=repo_url,
@@ -343,6 +395,7 @@ def validate_plugin(
             stage="timeout",
             message=f"worker timed out after {load_timeout} seconds",
             plugin_dir_name=plugin_dir_name,
+            details=build_process_output_details(stdout=exc.stdout, stderr=exc.stderr),
         )
 
     return parse_worker_output(
@@ -483,13 +536,10 @@ def run_worker(args: argparse.Namespace) -> int:
 
         os.environ["ASTRBOT_ROOT"] = str(astrbot_root)
         os.environ.setdefault("TESTING", "true")
-        for entry in reversed(
-            build_worker_sys_path(
-                astrbot_root=astrbot_root,
-                astrbot_path=Path(args.astrbot_path),
-            )
-        ):
-            sys.path.insert(0, entry)
+        sys.path[:0] = build_worker_sys_path(
+            astrbot_root=astrbot_root,
+            astrbot_path=Path(args.astrbot_path),
+        )
 
         result = asyncio.run(
             run_worker_load_check(args.plugin_dir_name, args.normalized_repo_url)
@@ -525,6 +575,7 @@ 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("--worker", action="store_true")
     parser.add_argument("--plugin-source-dir")
@@ -578,6 +629,7 @@ def main() -> int:
                 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

tests/test_validate_plugins.py

@@ -40,6 +40,39 @@ def test_rejects_non_github_urls(self):
         with self.assertRaises(ValueError):
             module.normalize_repo_url("https://gitlab.com/example/demo-plugin")
 
+    def test_rejects_non_http_schemes(self):
+        module = load_validator_module()
+
+        for url in (
+            "git://github.com/example/demo-plugin",
+            "ssh://github.com/example/demo-plugin",
+        ):
+            with self.subTest(url=url):
+                with self.assertRaisesRegex(ValueError, "repo URL must use http or https"):
+                    module.normalize_repo_url(url)
+
+    def test_rejects_missing_owner_or_repository(self):
+        module = load_validator_module()
+
+        for url in (
+            "https://github.com/",
+            "https://github.com/example",
+            "https://github.com/example/",
+            "https://github.com//demo-plugin",
+            "https://github.com/example//",
+        ):
+            with self.subTest(url=url):
+                with self.assertRaisesRegex(ValueError, "repo URL must include owner and repository"):
+                    module.normalize_repo_url(url)
+
+    def test_strips_leading_and_trailing_whitespace(self):
+        module = load_validator_module()
+
+        self.assertEqual(
+            module.normalize_repo_url("  https://github.com/example/demo-plugin  "),
+            "https://github.com/example/demo-plugin",
+        )
+
 
 class SelectPluginsTests(unittest.TestCase):
     def test_returns_all_plugins_when_limit_is_none(self):
@@ -88,6 +121,80 @@ def test_prefers_explicit_names_in_requested_order(self):
 
         self.assertEqual([item[0] for item in selected], ["plugin-c", "plugin-a"])
 
+    def test_respects_positive_limit_when_names_not_requested(self):
+        module = load_validator_module()
+        plugins = {
+            "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"},
+        }
+
+        selected = module.select_plugins(
+            plugins=plugins,
+            requested_names=None,
+            limit=1,
+        )
+
+        self.assertEqual([item[0] for item in selected], ["plugin-a"])
+
+    def test_raises_key_error_for_unknown_requested_plugin(self):
+        module = load_validator_module()
+        plugins = {
+            "known-plugin": {"repo": "https://github.com/example/known-plugin"},
+        }
+
+        with self.assertRaisesRegex(KeyError, "plugin not found: missing-plugin"):
+            module.select_plugins(
+                plugins=plugins,
+                requested_names=["known-plugin", "missing-plugin"],
+                limit=None,
+            )
+
+
+class HelperFunctionTests(unittest.TestCase):
+    def test_combine_requested_names_merges_trims_and_drops_empty_values(self):
+        module = load_validator_module()
+
+        combined = module.combine_requested_names(
+            plugin_names=["foo", "  bar  ", "", "   "],
+            plugin_name_list="baz,   qux  , ,foo ",
+        )
+
+        self.assertEqual(combined, ["foo", "bar", "baz", "qux", "foo"])
+
+    def test_combine_requested_names_handles_none_inputs(self):
+        module = load_validator_module()
+
+        self.assertEqual(module.combine_requested_names(None, None), [])
+
+    def test_sanitize_name_replaces_invalid_chars_and_falls_back_when_needed(self):
+        module = load_validator_module()
+
+        self.assertEqual(module.sanitize_name("  -invalid name!*?-  "), "invalid-name")
+        self.assertEqual(module.sanitize_name("valid-name_123"), "valid-name_123")
+        self.assertEqual(module.sanitize_name("   "), "plugin")
+        self.assertEqual(module.sanitize_name("!!!"), "plugin")
+
+    def test_build_plugin_clone_dir_is_unique_for_colliding_sanitized_names(self):
+        module = load_validator_module()
+
+        first = module.build_plugin_clone_dir(Path("/tmp/work"), "foo bar")
+        second = module.build_plugin_clone_dir(Path("/tmp/work"), "foo/bar")
+
+        self.assertNotEqual(first, second)
+        self.assertEqual(first.parent, Path("/tmp/work"))
+        self.assertEqual(second.parent, Path("/tmp/work"))
+
+    def test_build_process_output_details_keeps_partial_timeout_logs(self):
+        module = load_validator_module()
+
+        details = module.build_process_output_details(
+            stdout="line one\nline two\n",
+            stderr=b"warning\n",
+        )
+
+        self.assertEqual(details, {"stdout": "line one\nline two", "stderr": "warning"})
+
 
 class MetadataValidationTests(unittest.TestCase):
     def test_reports_missing_required_metadata_fields(self):