hermes-agent/tests/test_plugins.py

"""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.