hermes-agent/tests/hermes_cli/test_plugin_capabilities.py

"""Tests for the plugin capability model + consent flow (#64228).

Covers: declaration parsing, consent grant/persist, update re-consent on
added capabilities, fail-closed behavior on missing/corrupt consent state,
and backward compatibility with the legacy ``allow_*`` gates.
"""

from __future__ import annotations

from types import SimpleNamespace
from unittest.mock import MagicMock, patch

import pytest
import hermes_yaml as yaml

from hermes_cli.plugin_capabilities import (
    CAPABILITY_REGISTRY,
    capability_set_hash,
    consent_hash,
    declared_set_changed,
    granted_capabilities,
    parse_declared_capabilities,
    pending_capabilities,
    plugin_capability_granted,
    record_consent,
)


@pytest.fixture()
def hermes_home(tmp_path, monkeypatch):
    """Point HERMES_HOME at a tmp dir with an empty config.yaml."""
    monkeypatch.setenv("HERMES_HOME", str(tmp_path))
    (tmp_path / "config.yaml").write_text("{}\n", encoding="utf-8")
    return tmp_path


def _read_cfg(home):
    return yaml.safe_load((home / "config.yaml").read_text(encoding="utf-8")) or {}


# ── Registry sanity ──────────────────────────────────────────────────────────


class TestRegistry:
    def test_every_capability_has_legacy_gate(self):
        for spec in CAPABILITY_REGISTRY.values():
            assert spec.legacy_path, spec.id
            assert spec.description



# ── Declaration parsing ──────────────────────────────────────────────────────


class TestDeclarationParsing:

    def test_drops_unknown_ids(self):
        got = parse_declared_capabilities(["tools.override", "root.everything"])
        assert got == ["tools.override"]

    def test_non_list_ignored(self):
        assert parse_declared_capabilities("tools.override") == []
        assert parse_declared_capabilities({"a": 1}) == []
        assert parse_declared_capabilities(None) == []

    def test_non_string_entries_ignored(self):
        assert parse_declared_capabilities([1, None, "tools.override"]) == [
            "tools.override"
        ]

    def test_dedup_preserves_order(self):
        got = parse_declared_capabilities(
            ["llm.model_override", "tools.override", "llm.model_override"]
        )
        assert got == ["llm.model_override", "tools.override"]

    def test_manifest_field_lands_on_parsed_manifest(self, tmp_path):
        """PluginManifest picks up ``capabilities:`` from plugin.yaml."""
        from hermes_cli.plugins import parse_manifest_file

        plugin_dir = tmp_path / "capplug"
        plugin_dir.mkdir()
        (plugin_dir / "plugin.yaml").write_text(
            "name: capplug\nversion: '1.0'\n"
            "capabilities:\n  - tools.override\n  - bogus.capability\n",
            encoding="utf-8",
        )
        manifest = parse_manifest_file(
            plugin_dir / "plugin.yaml", plugin_dir, "user", ""
        )
        assert manifest is not None
        assert manifest.capabilities == ["tools.override"]

    def test_manifest_without_capabilities_field(self, tmp_path):
        from hermes_cli.plugins import parse_manifest_file

        plugin_dir = tmp_path / "plainplug"
        plugin_dir.mkdir()
        (plugin_dir / "plugin.yaml").write_text(
            "name: plainplug\n", encoding="utf-8"
        )
        manifest = parse_manifest_file(
            plugin_dir / "plugin.yaml", plugin_dir, "user", ""
        )
        assert manifest is not None
        assert manifest.capabilities == []

    def test_entrypoint_companion_metadata_declares_capabilities_without_import(
        self, monkeypatch
    ):
        """Installed plugins can declare consent metadata in dist entry points."""
        from hermes_cli import plugins as plugins_mod
        from hermes_cli.plugins import PluginManager

        load = MagicMock(side_effect=AssertionError("plugin code must not be imported"))
        plugin_ep = SimpleNamespace(
            name="thread-namer",
            value="thread_namer.plugin:register",
            group="hermes_agent.plugins",
            dist=SimpleNamespace(
                version="1.2.3",
                metadata={"Summary": "Names gateway threads"},
            ),
            load=load,
        )
        capability_ep = SimpleNamespace(
            name="thread-namer.gateway.platform_actions",
            value="thread_namer.plugin:register",
            group="hermes_agent.plugin_capabilities",
            load=load,
        )
        monkeypatch.setattr(
            plugins_mod.importlib.metadata,
            "entry_points",
            lambda: [plugin_ep, capability_ep],
        )

        manifests = PluginManager()._scan_entry_points()

        assert len(manifests) == 1
        assert manifests[0].name == "thread-namer"
        assert manifests[0].version == "1.2.3"
        assert manifests[0].description == "Names gateway threads"
        assert manifests[0].capabilities == ["gateway.platform_actions"]
        load.assert_not_called()


# ── Consent grant + persistence ──────────────────────────────────────────────


class TestConsentPersistence:
    def test_record_consent_persists_grant(self, hermes_home):
        record_consent("capplug", ["tools.override"], ["tools.override"])
        cfg = _read_cfg(hermes_home)
        entry = cfg["plugins"]["entries"]["capplug"]
        assert entry["granted_capabilities"] == ["tools.override"]
        assert entry["capabilities_consent"]["hash"] == capability_set_hash(
            ["tools.override"]
        )
        assert entry["capabilities_consent"]["granted_at"]
        # Bridge: legacy key mirrored so existing enforcement sites work.
        assert entry["allow_tool_override"] is True

    def test_record_consent_mirrors_nested_legacy_key(self, hermes_home):
        record_consent(
            "capplug", ["llm.model_override"], ["llm.model_override"]
        )
        entry = _read_cfg(hermes_home)["plugins"]["entries"]["capplug"]
        assert entry["llm"]["allow_model_override"] is True


    def test_grant_is_union_with_previous(self, hermes_home):
        record_consent("capplug", ["tools.override"], ["tools.override"])
        record_consent(
            "capplug",
            ["llm.model_override"],
            ["tools.override", "llm.model_override"],
        )
        assert granted_capabilities("capplug") == frozenset(
            {"tools.override", "llm.model_override"}
        )


    def test_declined_stays_off(self, hermes_home):
        # No record_consent call — nothing granted.
        assert plugin_capability_granted("capplug", "tools.override") is False
        assert pending_capabilities("capplug", ["tools.override"]) == [
            "tools.override"
        ]


# ── Update re-consent ────────────────────────────────────────────────────────


class TestUpdateReconsent:
    def test_added_capability_is_pending(self, hermes_home):
        # v1 declared + granted tools.override.
        record_consent("capplug", ["tools.override"], ["tools.override"])
        # v2 adds llm.model_override.
        declared_v2 = ["tools.override", "llm.model_override"]
        assert pending_capabilities("capplug", declared_v2) == [
            "llm.model_override"
        ]
        assert declared_set_changed("capplug", declared_v2) is True
        # The added capability stays ungranted until re-consent.
        assert plugin_capability_granted("capplug", "llm.model_override") is False
        # The previously granted one keeps working.
        assert plugin_capability_granted("capplug", "tools.override") is True

    def test_unchanged_set_needs_no_reconsent(self, hermes_home):
        record_consent("capplug", ["tools.override"], ["tools.override"])
        assert pending_capabilities("capplug", ["tools.override"]) == []
        assert declared_set_changed("capplug", ["tools.override"]) is False

    def test_hash_order_insensitive(self):
        a = capability_set_hash(["tools.override", "llm.model_override"])
        b = capability_set_hash(["llm.model_override", "tools.override"])
        assert a == b

    def test_no_consent_record_counts_as_changed(self, hermes_home):
        assert declared_set_changed("capplug", ["tools.override"]) is True

    def test_reconsent_grants_addition(self, hermes_home):
        record_consent("capplug", ["tools.override"], ["tools.override"])
        declared_v2 = ["tools.override", "llm.model_override"]
        record_consent(
            "capplug", pending_capabilities("capplug", declared_v2), declared_v2
        )
        assert plugin_capability_granted("capplug", "llm.model_override") is True
        assert declared_set_changed("capplug", declared_v2) is False


# ── Fail closed ──────────────────────────────────────────────────────────────


class TestFailClosed:
    def test_missing_config_not_granted(self, tmp_path, monkeypatch):
        monkeypatch.setenv("HERMES_HOME", str(tmp_path / "nonexistent"))
        assert plugin_capability_granted("capplug", "tools.override") is False
        assert granted_capabilities("capplug") == frozenset()

    def test_corrupt_granted_list_not_granted(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    capplug:\n"
            "      granted_capabilities: not-a-list\n",
            encoding="utf-8",
        )
        assert plugin_capability_granted("capplug", "tools.override") is False

    def test_corrupt_entry_types_not_granted(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    capplug: 42\n", encoding="utf-8"
        )
        assert plugin_capability_granted("capplug", "tools.override") is False

    def test_unknown_capability_denied(self, hermes_home):
        record_consent("capplug", ["tools.override"], ["tools.override"])
        assert plugin_capability_granted("capplug", "root.everything") is False

    def test_unknown_ids_in_granted_list_ignored(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    capplug:\n"
            "      granted_capabilities: [root.everything, 42]\n",
            encoding="utf-8",
        )
        assert granted_capabilities("capplug") == frozenset()
        assert plugin_capability_granted("capplug", "tools.override") is False

    def test_corrupt_consent_hash_counts_as_changed(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    capplug:\n"
            "      capabilities_consent: broken\n",
            encoding="utf-8",
        )
        assert consent_hash("capplug") is None
        assert declared_set_changed("capplug", ["tools.override"]) is True


# ── Legacy gate backward compat ──────────────────────────────────────────────


class TestLegacyGateCompat:
    def test_legacy_allow_tool_override_still_works(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    oldplug:\n"
            "      allow_tool_override: true\n",
            encoding="utf-8",
        )
        assert plugin_capability_granted("oldplug", "tools.override") is True

    def test_legacy_nested_llm_key_still_works(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    oldplug:\n"
            "      llm:\n        allow_model_override: true\n",
            encoding="utf-8",
        )
        assert plugin_capability_granted("oldplug", "llm.model_override") is True

    def test_legacy_false_stays_denied(self, hermes_home):
        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    oldplug:\n"
            "      allow_tool_override: false\n",
            encoding="utf-8",
        )
        assert plugin_capability_granted("oldplug", "tools.override") is False

    def test_tool_override_gate_uses_canonical_path(self, hermes_home):
        """PluginContext._tool_override_allowed honors capability grant."""
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        record_consent("capplug", ["tools.override"], ["tools.override"])
        manifest = PluginManifest(name="capplug", source="user", key="capplug")
        ctx = PluginContext(manifest, PluginManager())
        assert ctx._tool_override_allowed("write_file") is True

    def test_tool_override_gate_denies_without_grant(self, hermes_home):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        manifest = PluginManifest(name="capplug", source="user", key="capplug")
        ctx = PluginContext(manifest, PluginManager())
        assert ctx._tool_override_allowed("write_file") is False

    def test_tool_override_gate_legacy_key(self, hermes_home):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        (hermes_home / "config.yaml").write_text(
            "plugins:\n  entries:\n    oldplug:\n"
            "      allow_tool_override: true\n",
            encoding="utf-8",
        )
        manifest = PluginManifest(name="oldplug", source="user", key="oldplug")
        ctx = PluginContext(manifest, PluginManager())
        assert ctx._tool_override_allowed("write_file") is True

    def test_bundled_plugin_trusted(self, hermes_home):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        manifest = PluginManifest(name="bplug", source="bundled", key="bplug")
        ctx = PluginContext(manifest, PluginManager())
        assert ctx._tool_override_allowed("write_file") is True
        assert ctx.has_capability("tools.override") is True


# ── ctx.has_capability probing ───────────────────────────────────────────────


class TestHasCapability:
    def test_probe_granted(self, hermes_home):
        from hermes_cli.plugins import PluginContext, PluginManifest, PluginManager

        record_consent("capplug", ["llm.model_override"], ["llm.model_override"])
        manifest = PluginManifest(name="capplug", source="user", key="capplug")
        ctx = PluginContext(manifest, PluginManager())
        assert ctx.has_capability("llm.model_override") is True
        assert ctx.has_capability("tools.override") is False
        assert ctx.has_capability("nonsense.capability") is False


# ── Consent CLI flow ─────────────────────────────────────────────────────────


class TestConsentFlow:
    def _console(self, answers=None, interactive=True):
        console = MagicMock()
        it = iter(answers or [])
        console.input.side_effect = lambda *a, **k: next(it, "")
        return console

    def test_consent_yes_records_grant(self, hermes_home, monkeypatch):
        from hermes_cli.plugins_cmd import _run_capability_consent

        console = self._console(["y"])
        with patch("sys.stdin") as stdin, patch("sys.stdout") as stdout:
            stdin.isatty.return_value = True
            stdout.isatty.return_value = True
            granted = _run_capability_consent(
                console, "capplug", ["tools.override"], context="install"
            )
        assert granted is True
        assert plugin_capability_granted("capplug", "tools.override") is True

    def test_consent_decline_leaves_ungranted(self, hermes_home):
        from hermes_cli.plugins_cmd import _run_capability_consent

        console = self._console(["n"])
        with patch("sys.stdin") as stdin, patch("sys.stdout") as stdout:
            stdin.isatty.return_value = True
            stdout.isatty.return_value = True
            granted = _run_capability_consent(
                console, "capplug", ["tools.override"], context="install"
            )
        assert granted is False
        assert plugin_capability_granted("capplug", "tools.override") is False

    def test_non_interactive_fails_closed(self, hermes_home):
        from hermes_cli.plugins_cmd import _run_capability_consent

        console = self._console()
        with patch("sys.stdin") as stdin, patch("sys.stdout") as stdout:
            stdin.isatty.return_value = False
            stdout.isatty.return_value = False
            granted = _run_capability_consent(
                console, "capplug", ["tools.override"], context="install"
            )
        assert granted is False
        assert plugin_capability_granted("capplug", "tools.override") is False
        # No prompt was shown.
        console.input.assert_not_called()

    def test_already_granted_skips_prompt(self, hermes_home):
        from hermes_cli.plugins_cmd import _run_capability_consent

        record_consent("capplug", ["tools.override"], ["tools.override"])
        console = self._console()
        granted = _run_capability_consent(
            console, "capplug", ["tools.override"], context="enable"
        )
        assert granted is True
        console.input.assert_not_called()