Teknium
7a5062fbcd
Validate plugin manifests, imports, hook signatures, and runtime registrations through the real plugin loader in an isolated temporary home.
137 lines
4.3 KiB
Python
137 lines
4.3 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
from hermes_cli.subcommands.plugins import build_plugins_parser
|
|
|
|
|
|
def _parse_plugins_args(*argv: str):
|
|
parser = argparse.ArgumentParser()
|
|
subparsers = parser.add_subparsers(dest="command")
|
|
build_plugins_parser(subparsers, cmd_plugins=lambda args: None)
|
|
return parser.parse_args(["plugins", *argv])
|
|
|
|
|
|
def test_plugins_parser_exposes_doctor() -> None:
|
|
doctor = _parse_plugins_args("doctor", "sample", "--ci")
|
|
|
|
assert (doctor.plugins_action, doctor.target, doctor.ci) == (
|
|
"doctor",
|
|
"sample",
|
|
True,
|
|
)
|
|
|
|
|
|
def test_doctor_uses_registration_to_reject_bad_hook_and_callback_signature(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
from hermes_cli.plugin_dev import doctor_plugin
|
|
|
|
plugin = tmp_path / "bad-plugin"
|
|
plugin.mkdir()
|
|
(plugin / "plugin.yaml").write_text(
|
|
"\n".join(
|
|
[
|
|
"name: bad-plugin",
|
|
"version: 0.1.0",
|
|
"description: broken contract",
|
|
"provides_hooks:",
|
|
" - typo_hook",
|
|
" - pre_tool_call",
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
(plugin / "__init__.py").write_text(
|
|
"def callback(tool_name):\n"
|
|
" return None\n\n"
|
|
"def register(ctx):\n"
|
|
" ctx.register_hook('typo_hook', callback)\n"
|
|
" ctx.register_hook('pre_tool_call', callback)\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
report = doctor_plugin(plugin)
|
|
messages = "\n".join(f.message for f in report.findings)
|
|
assert report.ok is False
|
|
assert "unknown hook 'typo_hook'" in messages
|
|
assert "must accept **kwargs" in messages
|
|
|
|
|
|
def test_doctor_accepts_manifest_defaults_from_runtime_parser(tmp_path: Path) -> None:
|
|
from hermes_cli.plugin_dev import doctor_plugin
|
|
|
|
plugin = tmp_path / "minimal"
|
|
plugin.mkdir()
|
|
(plugin / "plugin.yaml").write_text("name: minimal\n", encoding="utf-8")
|
|
(plugin / "__init__.py").write_text(
|
|
"def register(ctx):\n pass\n", encoding="utf-8"
|
|
)
|
|
|
|
report = doctor_plugin(plugin)
|
|
assert report.ok, report.format_text()
|
|
assert report.manifest is not None
|
|
assert report.manifest.kind == "standalone"
|
|
|
|
|
|
def test_doctor_restores_global_tool_policy_and_module_state(tmp_path: Path) -> None:
|
|
import sys
|
|
|
|
from hermes_cli.plugin_dev import doctor_plugin
|
|
from tools.registry import registry
|
|
|
|
target = tmp_path / "cleanup-plugin"
|
|
target.mkdir()
|
|
(target / "plugin.yaml").write_text(
|
|
"name: cleanup-plugin\nprovides_tools: [cleanup_plugin_ping]\n",
|
|
encoding="utf-8",
|
|
)
|
|
(target / "__init__.py").write_text(
|
|
"import json\n\n"
|
|
"def ping(args, **kwargs):\n return json.dumps({'ok': True})\n\n"
|
|
"def register(ctx):\n"
|
|
" ctx.register_tool(name='cleanup_plugin_ping', toolset='cleanup', "
|
|
"schema={'name': 'cleanup_plugin_ping', 'description': 'test', "
|
|
"'parameters': {'type': 'object'}}, handler=ping)\n",
|
|
encoding="utf-8",
|
|
)
|
|
before_policy = dict(registry._plugin_override_policy)
|
|
before_modules = {
|
|
name
|
|
for name in sys.modules
|
|
if name == "hermes_plugins" or name.startswith("hermes_plugins.")
|
|
}
|
|
|
|
report = doctor_plugin(target)
|
|
|
|
assert report.ok, report.format_text()
|
|
assert report.registered_tools == ("cleanup_plugin_ping",)
|
|
assert registry.get_entry("cleanup_plugin_ping") is None
|
|
assert registry._plugin_override_policy == before_policy
|
|
after_modules = {
|
|
name
|
|
for name in sys.modules
|
|
if name == "hermes_plugins" or name.startswith("hermes_plugins.")
|
|
}
|
|
assert after_modules == before_modules
|
|
|
|
|
|
def test_doctor_blocks_live_network(tmp_path: Path) -> None:
|
|
from hermes_cli.plugin_dev import doctor_plugin
|
|
|
|
plugin = tmp_path / "network-plugin"
|
|
plugin.mkdir()
|
|
(plugin / "plugin.yaml").write_text("name: network-plugin\n", encoding="utf-8")
|
|
(plugin / "__init__.py").write_text(
|
|
"import socket\n\n"
|
|
"def register(ctx):\n"
|
|
" socket.create_connection(('example.com', 443))\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
report = doctor_plugin(plugin)
|
|
assert report.ok is False
|
|
assert "network access is disabled while Plugin Doctor runs" in report.format_text()
|