455bf2e853
The plugin system defined six lifecycle hooks but only pre_tool_call and
post_tool_call were invoked. This activates the remaining four so that
external plugins (e.g. memory systems) can hook into the conversation
loop without touching core code.
Hook semantics:
- on_session_start: fires once when a new session is created
- pre_llm_call: fires once per turn before the tool-calling loop;
plugins can return {"context": "..."} to inject into the ephemeral
system prompt (not cached, not persisted)
- post_llm_call: fires once per turn after the loop completes, with
user_message and assistant_response for sync/storage
- on_session_end: fires at the end of every run_conversation call
invoke_hook() now returns a list of non-None callback return values,
enabling pre_llm_call context injection while remaining backward
compatible (existing hooks that return None are unaffected).
Salvaged from PR #2823.
Co-authored-by: Nicolò Boschi <boschi1997@gmail.com>
410 lines
16 KiB
Python
410 lines
16 KiB
Python
"""Tests for the Hermes plugin system (hermes_cli.plugins)."""
|
||
|
||
import logging
|
||
import os
|
||
import sys
|
||
import types
|
||
from pathlib import Path
|
||
from unittest.mock import MagicMock, patch
|
||
|
||
import pytest
|
||
import yaml
|
||
|
||
from hermes_cli.plugins import (
|
||
ENTRY_POINTS_GROUP,
|
||
VALID_HOOKS,
|
||
LoadedPlugin,
|
||
PluginContext,
|
||
PluginManager,
|
||
PluginManifest,
|
||
get_plugin_manager,
|
||
get_plugin_tool_names,
|
||
discover_plugins,
|
||
invoke_hook,
|
||
)
|
||
|
||
|
||
# ── Helpers ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _make_plugin_dir(base: Path, name: str, *, register_body: str = "pass",
|
||
manifest_extra: dict | None = None) -> Path:
|
||
"""Create a minimal plugin directory with plugin.yaml + __init__.py."""
|
||
plugin_dir = base / name
|
||
plugin_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
manifest = {"name": name, "version": "0.1.0", "description": f"Test plugin {name}"}
|
||
if manifest_extra:
|
||
manifest.update(manifest_extra)
|
||
|
||
(plugin_dir / "plugin.yaml").write_text(yaml.dump(manifest))
|
||
(plugin_dir / "__init__.py").write_text(
|
||
f"def register(ctx):\n {register_body}\n"
|
||
)
|
||
return plugin_dir
|
||
|
||
|
||
# ── TestPluginDiscovery ────────────────────────────────────────────────────
|
||
|
||
|
||
class TestPluginDiscovery:
|
||
"""Tests for plugin discovery from directories and entry points."""
|
||
|
||
def test_discover_user_plugins(self, tmp_path, monkeypatch):
|
||
"""Plugins in ~/.hermes/plugins/ are discovered."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "hello_plugin")
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "hello_plugin" in mgr._plugins
|
||
assert mgr._plugins["hello_plugin"].enabled
|
||
|
||
def test_discover_project_plugins(self, tmp_path, monkeypatch):
|
||
"""Plugins in ./.hermes/plugins/ are discovered."""
|
||
project_dir = tmp_path / "project"
|
||
project_dir.mkdir()
|
||
monkeypatch.chdir(project_dir)
|
||
monkeypatch.setenv("HERMES_ENABLE_PROJECT_PLUGINS", "true")
|
||
plugins_dir = project_dir / ".hermes" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "proj_plugin")
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "proj_plugin" in mgr._plugins
|
||
assert mgr._plugins["proj_plugin"].enabled
|
||
|
||
def test_discover_project_plugins_skipped_by_default(self, tmp_path, monkeypatch):
|
||
"""Project plugins are not discovered unless explicitly enabled."""
|
||
project_dir = tmp_path / "project"
|
||
project_dir.mkdir()
|
||
monkeypatch.chdir(project_dir)
|
||
plugins_dir = project_dir / ".hermes" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "proj_plugin")
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "proj_plugin" not in mgr._plugins
|
||
|
||
def test_discover_is_idempotent(self, tmp_path, monkeypatch):
|
||
"""Calling discover_and_load() twice does not duplicate plugins."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "once_plugin")
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
mgr.discover_and_load() # second call should no-op
|
||
|
||
assert len(mgr._plugins) == 1
|
||
|
||
def test_discover_skips_dir_without_manifest(self, tmp_path, monkeypatch):
|
||
"""Directories without plugin.yaml are silently skipped."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
(plugins_dir / "no_manifest").mkdir(parents=True)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert len(mgr._plugins) == 0
|
||
|
||
def test_entry_points_scanned(self, tmp_path, monkeypatch):
|
||
"""Entry-point based plugins are discovered (mocked)."""
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
fake_module = types.ModuleType("fake_ep_plugin")
|
||
fake_module.register = lambda ctx: None # type: ignore[attr-defined]
|
||
|
||
fake_ep = MagicMock()
|
||
fake_ep.name = "ep_plugin"
|
||
fake_ep.value = "fake_ep_plugin:register"
|
||
fake_ep.group = ENTRY_POINTS_GROUP
|
||
fake_ep.load.return_value = fake_module
|
||
|
||
def fake_entry_points():
|
||
result = MagicMock()
|
||
result.select = MagicMock(return_value=[fake_ep])
|
||
return result
|
||
|
||
with patch("importlib.metadata.entry_points", fake_entry_points):
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "ep_plugin" in mgr._plugins
|
||
|
||
|
||
# ── TestPluginLoading ──────────────────────────────────────────────────────
|
||
|
||
|
||
class TestPluginLoading:
|
||
"""Tests for plugin module loading."""
|
||
|
||
def test_load_missing_init(self, tmp_path, monkeypatch):
|
||
"""Plugin dir without __init__.py records an error."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
plugin_dir = plugins_dir / "bad_plugin"
|
||
plugin_dir.mkdir(parents=True)
|
||
(plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "bad_plugin"}))
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "bad_plugin" in mgr._plugins
|
||
assert not mgr._plugins["bad_plugin"].enabled
|
||
assert mgr._plugins["bad_plugin"].error is not None
|
||
|
||
def test_load_missing_register_fn(self, tmp_path, monkeypatch):
|
||
"""Plugin without register() function records an error."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
plugin_dir = plugins_dir / "no_reg"
|
||
plugin_dir.mkdir(parents=True)
|
||
(plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "no_reg"}))
|
||
(plugin_dir / "__init__.py").write_text("# no register function\n")
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "no_reg" in mgr._plugins
|
||
assert not mgr._plugins["no_reg"].enabled
|
||
assert "no register()" in mgr._plugins["no_reg"].error
|
||
|
||
def test_load_registers_namespace_module(self, tmp_path, monkeypatch):
|
||
"""Directory plugins are importable under hermes_plugins.<name>."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "ns_plugin")
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
# Clean up any prior namespace module
|
||
sys.modules.pop("hermes_plugins.ns_plugin", None)
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "hermes_plugins.ns_plugin" in sys.modules
|
||
|
||
|
||
# ── TestPluginHooks ────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestPluginHooks:
|
||
"""Tests for lifecycle hook registration and invocation."""
|
||
|
||
def test_register_and_invoke_hook(self, tmp_path, monkeypatch):
|
||
"""Registered hooks are called on invoke_hook()."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(
|
||
plugins_dir, "hook_plugin",
|
||
register_body='ctx.register_hook("pre_tool_call", lambda **kw: None)',
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
# Should not raise
|
||
mgr.invoke_hook("pre_tool_call", tool_name="test", args={}, task_id="t1")
|
||
|
||
def test_hook_exception_does_not_propagate(self, tmp_path, monkeypatch):
|
||
"""A hook callback that raises does NOT crash the caller."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(
|
||
plugins_dir, "bad_hook",
|
||
register_body='ctx.register_hook("post_tool_call", lambda **kw: 1/0)',
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
# Should not raise despite 1/0
|
||
mgr.invoke_hook("post_tool_call", tool_name="x", args={}, result="r", task_id="")
|
||
|
||
def test_hook_return_values_collected(self, tmp_path, monkeypatch):
|
||
"""invoke_hook() collects non-None return values from callbacks."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(
|
||
plugins_dir, "ctx_plugin",
|
||
register_body=(
|
||
'ctx.register_hook("pre_llm_call", '
|
||
'lambda **kw: {"context": "memory from plugin"})'
|
||
),
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
results = mgr.invoke_hook("pre_llm_call", session_id="s1", user_message="hi",
|
||
conversation_history=[], is_first_turn=True, model="test")
|
||
assert len(results) == 1
|
||
assert results[0] == {"context": "memory from plugin"}
|
||
|
||
def test_hook_none_returns_excluded(self, tmp_path, monkeypatch):
|
||
"""invoke_hook() excludes None returns from the result list."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(
|
||
plugins_dir, "none_hook",
|
||
register_body='ctx.register_hook("post_llm_call", lambda **kw: None)',
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
results = mgr.invoke_hook("post_llm_call", session_id="s1",
|
||
user_message="hi", assistant_response="bye", model="test")
|
||
assert results == []
|
||
|
||
def test_invalid_hook_name_warns(self, tmp_path, monkeypatch, caplog):
|
||
"""Registering an unknown hook name logs a warning."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(
|
||
plugins_dir, "warn_plugin",
|
||
register_body='ctx.register_hook("on_banana", lambda **kw: None)',
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
with caplog.at_level(logging.WARNING, logger="hermes_cli.plugins"):
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert any("on_banana" in record.message for record in caplog.records)
|
||
|
||
|
||
# ── TestPluginContext ──────────────────────────────────────────────────────
|
||
|
||
|
||
class TestPluginContext:
|
||
"""Tests for the PluginContext facade."""
|
||
|
||
def test_register_tool_adds_to_registry(self, tmp_path, monkeypatch):
|
||
"""PluginContext.register_tool() puts the tool in the global registry."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
plugin_dir = plugins_dir / "tool_plugin"
|
||
plugin_dir.mkdir(parents=True)
|
||
(plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "tool_plugin"}))
|
||
(plugin_dir / "__init__.py").write_text(
|
||
'def register(ctx):\n'
|
||
' ctx.register_tool(\n'
|
||
' name="plugin_echo",\n'
|
||
' toolset="plugin_tool_plugin",\n'
|
||
' schema={"name": "plugin_echo", "description": "Echo", "parameters": {"type": "object", "properties": {}}},\n'
|
||
' handler=lambda args, **kw: "echo",\n'
|
||
' )\n'
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
assert "plugin_echo" in mgr._plugin_tool_names
|
||
|
||
from tools.registry import registry
|
||
assert "plugin_echo" in registry._tools
|
||
|
||
|
||
# ── TestPluginToolVisibility ───────────────────────────────────────────────
|
||
|
||
|
||
class TestPluginToolVisibility:
|
||
"""Plugin-registered tools appear in get_tool_definitions()."""
|
||
|
||
def test_plugin_tools_in_definitions(self, tmp_path, monkeypatch):
|
||
"""Plugin tools are included when their toolset is in enabled_toolsets."""
|
||
import hermes_cli.plugins as plugins_mod
|
||
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
plugin_dir = plugins_dir / "vis_plugin"
|
||
plugin_dir.mkdir(parents=True)
|
||
(plugin_dir / "plugin.yaml").write_text(yaml.dump({"name": "vis_plugin"}))
|
||
(plugin_dir / "__init__.py").write_text(
|
||
'def register(ctx):\n'
|
||
' ctx.register_tool(\n'
|
||
' name="vis_tool",\n'
|
||
' toolset="plugin_vis_plugin",\n'
|
||
' schema={"name": "vis_tool", "description": "Visible", "parameters": {"type": "object", "properties": {}}},\n'
|
||
' handler=lambda args, **kw: "ok",\n'
|
||
' )\n'
|
||
)
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
monkeypatch.setattr(plugins_mod, "_plugin_manager", mgr)
|
||
|
||
from model_tools import get_tool_definitions
|
||
|
||
# Plugin tools are included when their toolset is explicitly enabled
|
||
tools = get_tool_definitions(enabled_toolsets=["terminal", "plugin_vis_plugin"], quiet_mode=True)
|
||
tool_names = [t["function"]["name"] for t in tools]
|
||
assert "vis_tool" in tool_names
|
||
|
||
# Plugin tools are excluded when only other toolsets are enabled
|
||
tools2 = get_tool_definitions(enabled_toolsets=["terminal"], quiet_mode=True)
|
||
tool_names2 = [t["function"]["name"] for t in tools2]
|
||
assert "vis_tool" not in tool_names2
|
||
|
||
# Plugin tools are included when no toolset filter is active (all enabled)
|
||
tools3 = get_tool_definitions(quiet_mode=True)
|
||
tool_names3 = [t["function"]["name"] for t in tools3]
|
||
assert "vis_tool" in tool_names3
|
||
|
||
|
||
# ── TestPluginManagerList ──────────────────────────────────────────────────
|
||
|
||
|
||
class TestPluginManagerList:
|
||
"""Tests for PluginManager.list_plugins()."""
|
||
|
||
def test_list_empty(self):
|
||
"""Empty manager returns empty list."""
|
||
mgr = PluginManager()
|
||
assert mgr.list_plugins() == []
|
||
|
||
def test_list_returns_sorted(self, tmp_path, monkeypatch):
|
||
"""list_plugins() returns results sorted by name."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "zulu")
|
||
_make_plugin_dir(plugins_dir, "alpha")
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
listing = mgr.list_plugins()
|
||
names = [p["name"] for p in listing]
|
||
assert names == sorted(names)
|
||
|
||
def test_list_with_plugins(self, tmp_path, monkeypatch):
|
||
"""list_plugins() returns info dicts for each discovered plugin."""
|
||
plugins_dir = tmp_path / "hermes_test" / "plugins"
|
||
_make_plugin_dir(plugins_dir, "alpha")
|
||
_make_plugin_dir(plugins_dir, "beta")
|
||
monkeypatch.setenv("HERMES_HOME", str(tmp_path / "hermes_test"))
|
||
|
||
mgr = PluginManager()
|
||
mgr.discover_and_load()
|
||
|
||
listing = mgr.list_plugins()
|
||
names = [p["name"] for p in listing]
|
||
assert "alpha" in names
|
||
assert "beta" in names
|
||
for p in listing:
|
||
assert "enabled" in p
|
||
assert "tools" in p
|
||
assert "hooks" in p
|
||
|
||
|
||
|
||
# NOTE: TestPluginCommands removed – register_command() was never implemented
|
||
# in PluginContext (hermes_cli/plugins.py). The tests referenced _plugin_commands,
|
||
# commands_registered, get_plugin_command_handler, and GATEWAY_KNOWN_COMMANDS
|
||
# integration — all of which are unimplemented features.
|