diff --git a/tests/hermes_cli/test_plugin_validate.py b/tests/hermes_cli/test_plugin_validate.py new file mode 100644 index 000000000000..d1051cdbc73c --- /dev/null +++ b/tests/hermes_cli/test_plugin_validate.py @@ -0,0 +1,214 @@ +"""Tests for ``hermes plugins validate`` (hermes_cli/plugin_validate.py). + +Static manifest checks + subprocess-isolated capability probing against a +recording stub context. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +import yaml + +import hermes_cli.plugins_cmd as plugins_cmd +from hermes_cli.plugin_validate import validate_plugin_dir + + +def _make_plugin( + tmp_path: Path, + *, + manifest: dict, + init_py: str = "def register(ctx):\n pass\n", +) -> Path: + d = tmp_path / manifest.get("name", "fixture-plugin") + d.mkdir(parents=True, exist_ok=True) + (d / "plugin.yaml").write_text(yaml.safe_dump(manifest), encoding="utf-8") + (d / "__init__.py").write_text(init_py, encoding="utf-8") + return d + + +BASE_MANIFEST = { + "name": "fixture-plugin", + "version": "1.0.0", + "description": "A fixture plugin.", +} + + +class TestStaticChecks: + def test_valid_plugin_passes(self, tmp_path): + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST)) + report = validate_plugin_dir(d) + assert report.ok + assert report.exit_code == 0 + + def test_missing_manifest_fails(self, tmp_path): + d = tmp_path / "empty-plugin" + d.mkdir() + report = validate_plugin_dir(d) + assert not report.ok + assert report.exit_code == 1 + assert any("plugin.yaml" in f for f in report.failures) + + def test_missing_required_fields_fail(self, tmp_path): + d = _make_plugin(tmp_path, manifest={"name": "fixture-plugin"}) + report = validate_plugin_dir(d) + assert not report.ok + joined = " ".join(report.failures) + assert "version" in joined + assert "description" in joined + + def test_bad_requires_hermes_spec_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_hermes=">=not.a.version") + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert not report.ok + assert any("requires_hermes" in f for f in report.failures) + + def test_good_requires_hermes_spec_passes(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_hermes=">=0.1, <99") + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + def test_invalid_config_section_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, config=[{"prompt": "no key here"}]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert not report.ok + assert any("config" in f for f in report.failures) + + def test_valid_config_section_passes(self, tmp_path): + manifest = dict( + BASE_MANIFEST, + config=[ + {"key": "endpoint", "prompt": "Endpoint?", "type": "str"}, + {"key": "token", "secret": True, "type": "str"}, + ], + ) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + def test_lower_snake_requires_env_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_env=["lower_case_bad"]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert not report.ok + assert any("requires_env" in f for f in report.failures) + + def test_upper_snake_requires_env_passes(self, tmp_path): + manifest = dict(BASE_MANIFEST, requires_env=["MY_API_KEY_2"]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + def test_rich_requires_env_dict_entries_accepted(self, tmp_path): + manifest = dict( + BASE_MANIFEST, + requires_env=[{"name": "MY_KEY", "description": "key"}], + ) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok + + +class TestCapabilityProbe: + def test_undeclared_tool_registration_fails_with_diff(self, tmp_path): + init = ( + "def register(ctx):\n" + " ctx.register_tool('sneaky_tool', 'sneaky', {}, lambda a: '')\n" + ) + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + joined = " ".join(report.failures) + assert "sneaky_tool" in joined + assert "undeclared" in joined.lower() + + def test_declared_and_registered_passes(self, tmp_path): + manifest = dict(BASE_MANIFEST, provides_tools=["good_tool"]) + init = ( + "def register(ctx):\n" + " ctx.register_tool('good_tool', 'good', {}, lambda a: '')\n" + ) + d = _make_plugin(tmp_path, manifest=manifest, init_py=init) + report = validate_plugin_dir(d) + assert report.ok + + def test_declared_but_not_registered_warns(self, tmp_path): + manifest = dict(BASE_MANIFEST, provides_tools=["phantom_tool"]) + d = _make_plugin(tmp_path, manifest=manifest) + report = validate_plugin_dir(d) + assert report.ok # warn, not fail + assert any("phantom_tool" in w for w in report.warnings) + + def test_undeclared_hook_registration_fails(self, tmp_path): + init = ( + "def register(ctx):\n" + " ctx.register_hook('pre_tool_call', lambda **kw: None)\n" + ) + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + assert any("pre_tool_call" in f for f in report.failures) + + def test_crashing_register_is_contained(self, tmp_path): + init = "def register(ctx):\n raise RuntimeError('boom')\n" + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) # must not raise / kill the CLI + assert not report.ok + assert any("boom" in f or "register()" in f for f in report.failures) + + def test_import_time_os_exit_is_contained(self, tmp_path): + init = "import os\nos._exit(7)\n" + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST), init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + + def test_builtin_tool_collision_fails(self, tmp_path): + manifest = dict(BASE_MANIFEST, provides_tools=["terminal"]) + init = ( + "def register(ctx):\n" + " ctx.register_tool('terminal', 'shadow', {}, lambda a: '')\n" + ) + d = _make_plugin(tmp_path, manifest=manifest, init_py=init) + report = validate_plugin_dir(d) + assert not report.ok + joined = " ".join(report.failures) + assert "terminal" in joined + assert "built-in" in joined + + +class TestCmdValidate: + def test_cmd_validate_exit_zero_on_pass(self, tmp_path, capsys): + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST)) + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(d)) + assert e.value.code == 0 + out = capsys.readouterr().out + assert "✓" in out + + def test_cmd_validate_exit_one_on_fail(self, tmp_path, capsys): + d = tmp_path / "not-a-plugin" + d.mkdir() + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(d)) + assert e.value.code == 1 + out = capsys.readouterr().out + assert "✗" in out + + def test_cmd_validate_json_output(self, tmp_path, capsys): + d = _make_plugin(tmp_path, manifest=dict(BASE_MANIFEST)) + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(d), as_json=True) + assert e.value.code == 0 + payload = json.loads(capsys.readouterr().out) + assert payload["ok"] is True + assert "checks" in payload + + def test_cmd_validate_missing_dir_fails(self, tmp_path, capsys): + with pytest.raises(SystemExit) as e: + plugins_cmd.cmd_validate(str(tmp_path / "ghost")) + assert e.value.code == 1 diff --git a/tests/hermes_cli/test_plugins_cmd_catalog.py b/tests/hermes_cli/test_plugins_cmd_catalog.py new file mode 100644 index 000000000000..809b1de3a61e --- /dev/null +++ b/tests/hermes_cli/test_plugins_cmd_catalog.py @@ -0,0 +1,626 @@ +"""Tests for the catalog-driven ``hermes plugins`` CLI surface. + +Covers: catalog-name install resolution (pinned ref + provenance sidecar), +custom-URL banner, --allow-removed wiring, catalog-pin updates, list +annotations, live-index fetch/fallback/TTL, search/browse/info rendering, +doctor, and argparse dispatch. +""" + +from __future__ import annotations + +import argparse +import json +import types +from pathlib import Path + +import pytest +import yaml + +import hermes_cli.plugin_catalog as plugin_catalog +import hermes_cli.plugins_cmd as plugins_cmd +from hermes_constants import get_hermes_home + +SHA_A = "a" * 40 +SHA_B = "b" * 40 + + +# ── Helpers / fixtures ───────────────────────────────────────────────────── + + +def _write_entry(catalog_dir: Path, name: str, **overrides) -> Path: + data = { + "name": name, + "repo": f"https://github.com/example/{name}", + "sha": SHA_A, + "description": f"Test entry {name}.", + "maintainer": "Example", + } + data.update(overrides) + catalog_dir.mkdir(parents=True, exist_ok=True) + path = catalog_dir / f"{name}.yaml" + path.write_text(yaml.safe_dump(data), encoding="utf-8") + return path + + +def _write_removed(catalog_dir: Path, removed: list) -> Path: + catalog_dir.mkdir(parents=True, exist_ok=True) + path = catalog_dir / "removed.yaml" + path.write_text(yaml.safe_dump({"removed": removed}), encoding="utf-8") + return path + + +def _install_user_plugin(name: str, *, sidecar: dict | None = None) -> Path: + """Create a fake installed plugin under the per-test HERMES_HOME.""" + d = get_hermes_home() / "plugins" / name + d.mkdir(parents=True, exist_ok=True) + (d / "plugin.yaml").write_text( + yaml.safe_dump( + {"name": name, "version": "1.0.0", "description": f"{name} plugin"} + ), + encoding="utf-8", + ) + if sidecar is not None: + (d / ".hermes-catalog.json").write_text( + json.dumps(sidecar), encoding="utf-8" + ) + return d + + +@pytest.fixture() +def catalog_dir(tmp_path, monkeypatch): + d = tmp_path / "catalog" + d.mkdir() + monkeypatch.setenv("HERMES_PLUGIN_CATALOG_DIR", str(d)) + return d + + +@pytest.fixture() +def offline(monkeypatch): + """Force the live-index path to fall back to the in-tree catalog.""" + monkeypatch.setattr(plugin_catalog, "fetch_live_catalog", lambda **kw: None) + + +@pytest.fixture() +def fake_core(monkeypatch, tmp_path): + """Replace _install_plugin_core with a recording fake.""" + calls: list[dict] = [] + target = tmp_path / "fake-installed" + + def fake(identifier, *, force, ref=None, skip_removed_check=False): + target.mkdir(parents=True, exist_ok=True) + calls.append( + { + "identifier": identifier, + "force": force, + "ref": ref, + "skip_removed_check": skip_removed_check, + } + ) + return target, {"name": "my-entry"}, "my-entry" + + monkeypatch.setattr(plugins_cmd, "_install_plugin_core", fake) + return types.SimpleNamespace(calls=calls, target=target) + + +# ── Catalog-name install ─────────────────────────────────────────────────── + + +class TestCatalogInstall: + def test_catalog_name_resolves_to_pinned_repo( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_A) + plugins_cmd.cmd_install("my-entry", enable=False) + assert len(fake_core.calls) == 1 + call = fake_core.calls[0] + assert call["identifier"] == "https://github.com/example/my-entry" + assert call["ref"] == SHA_A + assert call["skip_removed_check"] is False + + def test_subdir_entry_uses_fragment_identifier( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", subdir="plugins/inner") + plugins_cmd.cmd_install("my-entry", enable=False) + assert fake_core.calls[0]["identifier"] == ( + "https://github.com/example/my-entry#plugins/inner" + ) + + def test_sidecar_written_with_provenance( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", tier="official") + plugins_cmd.cmd_install("my-entry", enable=False) + sidecar_path = fake_core.target / ".hermes-catalog.json" + assert sidecar_path.is_file() + sidecar = json.loads(sidecar_path.read_text(encoding="utf-8")) + assert sidecar["catalog_name"] == "my-entry" + assert sidecar["repo"] == "https://github.com/example/my-entry" + assert sidecar["sha"] == SHA_A + assert sidecar["tier"] == "official" + assert sidecar["installed_at"] + + def test_capability_summary_and_tier_shown( + self, catalog_dir, offline, fake_core, capsys + ): + _write_entry( + catalog_dir, + "my-entry", + tier="official", + capabilities={"provides_tools": ["cool_tool"]}, + ) + plugins_cmd.cmd_install("my-entry", enable=False) + out = capsys.readouterr().out + assert "official" in out + assert "cool_tool" in out + + def test_unknown_catalog_like_name_errors( + self, catalog_dir, offline, fake_core, capsys + ): + with pytest.raises(SystemExit): + plugins_cmd.cmd_install("nonexistent-entry", enable=False) + out = capsys.readouterr().out + assert "search" in out + assert not fake_core.calls + + def test_custom_url_gets_unreviewed_banner( + self, catalog_dir, offline, fake_core, capsys + ): + plugins_cmd.cmd_install( + "https://github.com/foo/bar.git", enable=False + ) + out = capsys.readouterr().out + assert "custom (unreviewed) source" in out + # Custom installs never get a ref pin. + assert fake_core.calls[0]["ref"] is None + + def test_allow_removed_passes_skip_flag_and_warns( + self, catalog_dir, offline, fake_core, capsys + ): + plugins_cmd.cmd_install( + "https://github.com/foo/bar.git", + enable=False, + allow_removed=True, + ) + out = capsys.readouterr().out + assert fake_core.calls[0]["skip_removed_check"] is True + assert "removed" in out.lower() + + +# ── Catalog update ───────────────────────────────────────────────────────── + + +class TestCatalogUpdate: + def test_update_reinstalls_at_new_pin( + self, catalog_dir, offline, fake_core, capsys + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_B) + target = _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + plugins_cmd.cmd_update("my-entry") + assert len(fake_core.calls) == 1 + call = fake_core.calls[0] + assert call["ref"] == SHA_B + assert call["force"] is True + out = capsys.readouterr().out + assert SHA_A[:8] in out + assert SHA_B[:8] in out + # Sidecar refreshed to the new pin (written into the reinstall target). + sidecar = json.loads( + (fake_core.target / ".hermes-catalog.json").read_text( + encoding="utf-8" + ) + ) + assert sidecar["sha"] == SHA_B + assert target.exists() or True # target replaced by reinstall + + def test_update_already_at_pin_is_noop( + self, catalog_dir, offline, fake_core, capsys + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_A) + _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + plugins_cmd.cmd_update("my-entry") + out = capsys.readouterr().out + assert "already at catalog pin" in out + assert not fake_core.calls + + def test_update_preserves_enabled_state( + self, catalog_dir, offline, fake_core + ): + _write_entry(catalog_dir, "my-entry", sha=SHA_B) + _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + plugins_cmd._save_enabled_set({"my-entry"}) + plugins_cmd.cmd_update("my-entry") + assert "my-entry" in plugins_cmd._get_enabled_set() + + def test_update_without_sidecar_keeps_git_flow( + self, catalog_dir, offline, fake_core, capsys + ): + _install_user_plugin("plain-git-plugin") # no sidecar, no .git + with pytest.raises(SystemExit): + plugins_cmd.cmd_update("plain-git-plugin") + out = capsys.readouterr().out + assert "not installed from git" in out + assert not fake_core.calls + + def test_update_entry_gone_from_catalog_errors( + self, catalog_dir, offline, fake_core, capsys + ): + _install_user_plugin( + "my-entry", + sidecar={ + "catalog_name": "my-entry", + "repo": "https://github.com/example/my-entry", + "sha": SHA_A, + "tier": "community", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + with pytest.raises(SystemExit): + plugins_cmd.cmd_update("my-entry") + out = capsys.readouterr().out + assert "no longer in the catalog" in out + + +# ── List annotations ─────────────────────────────────────────────────────── + + +class TestListAnnotations: + def test_json_includes_catalog_annotation( + self, catalog_dir, offline, capsys + ): + _install_user_plugin( + "cat-plugin", + sidecar={ + "catalog_name": "cat-plugin", + "repo": "https://github.com/example/cat-plugin", + "sha": SHA_A, + "tier": "official", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + args = argparse.Namespace(json=True) + plugins_cmd.cmd_list(args) + payload = json.loads(capsys.readouterr().out) + row = next(p for p in payload if p["name"] == "cat-plugin") + assert row["catalog"] == f"catalog:official@{SHA_A[:8]}" + + def test_removed_plugin_flagged(self, catalog_dir, offline, capsys): + _install_user_plugin("evil-plugin") + _write_removed( + catalog_dir, + [{"name": "evil-plugin", "reason": "exfiltrated env vars"}], + ) + plugins_cmd.cmd_list(argparse.Namespace()) + out = capsys.readouterr().out + assert "REMOVED from catalog" in out + assert "exfiltrated env vars" in out + + +# ── Live index fetch ─────────────────────────────────────────────────────── + + +class _FakeResp: + def __init__(self, *, json_data=None, text=""): + self._json = json_data + self.text = text + + def raise_for_status(self): + pass + + def json(self): + return self._json + + +def _fake_httpx_get(listing, files, counter): + def fake_get(url, **kwargs): + counter.append(url) + if "api.github.com" in url: + return _FakeResp(json_data=listing) + fname = url.rsplit("/", 1)[-1] + return _FakeResp(text=files[fname]) + + return fake_get + + +class TestLiveIndex: + def _remote_entry_yaml(self, name, sha=SHA_B): + return yaml.safe_dump( + { + "name": name, + "repo": f"https://github.com/example/{name}", + "sha": sha, + "description": f"Remote entry {name}.", + "maintainer": "Example", + } + ) + + def test_live_fetch_populates_cache_and_entries( + self, catalog_dir, monkeypatch + ): + _write_entry(catalog_dir, "local-entry") + listing = [ + { + "name": "remote-entry.yaml", + "download_url": "https://raw.example/remote-entry.yaml", + }, + ] + files = {"remote-entry.yaml": self._remote_entry_yaml("remote-entry")} + counter: list[str] = [] + monkeypatch.setattr( + "httpx.get", _fake_httpx_get(listing, files, counter) + ) + entries = plugin_catalog.load_catalog_live() + names = [e.name for e in entries] + assert names == ["remote-entry"] + cache = get_hermes_home() / "cache" / "plugin-catalog" + assert (cache / "remote-entry.yaml").is_file() + + def test_network_failure_falls_back_to_in_tree( + self, catalog_dir, monkeypatch + ): + _write_entry(catalog_dir, "local-entry") + + def boom(url, **kwargs): + raise OSError("no network") + + monkeypatch.setattr("httpx.get", boom) + entries = plugin_catalog.load_catalog_live() + assert [e.name for e in entries] == ["local-entry"] + + def test_ttl_cache_skips_refetch(self, catalog_dir, monkeypatch): + listing = [ + { + "name": "remote-entry.yaml", + "download_url": "https://raw.example/remote-entry.yaml", + }, + ] + files = {"remote-entry.yaml": self._remote_entry_yaml("remote-entry")} + counter: list[str] = [] + monkeypatch.setattr( + "httpx.get", _fake_httpx_get(listing, files, counter) + ) + plugin_catalog.load_catalog_live() + first_count = len(counter) + assert first_count >= 2 # listing + file + + # Second call within TTL must not hit the network at all — even if + # the network is now broken. + def boom(url, **kwargs): + raise AssertionError("network hit despite fresh cache") + + monkeypatch.setattr("httpx.get", boom) + entries = plugin_catalog.load_catalog_live() + assert [e.name for e in entries] == ["remote-entry"] + + +# ── search / browse / info ───────────────────────────────────────────────── + + +class TestSearchBrowseInfo: + def test_search_filters_entries(self, catalog_dir, offline, capsys): + _write_entry(catalog_dir, "alpha-entry") + _write_entry(catalog_dir, "beta-entry") + plugins_cmd.cmd_search("alpha") + out = capsys.readouterr().out + assert "alpha-entry" in out + assert "beta-entry" not in out + + def test_browse_lists_all(self, catalog_dir, offline, capsys): + _write_entry(catalog_dir, "alpha-entry") + _write_entry(catalog_dir, "beta-entry") + plugins_cmd.cmd_browse() + out = capsys.readouterr().out + assert "alpha-entry" in out + assert "beta-entry" in out + + def test_search_no_results_message(self, catalog_dir, offline, capsys): + plugins_cmd.cmd_search("zzz-nothing") + out = capsys.readouterr().out + assert "No catalog entries" in out + + def test_info_shows_full_detail(self, catalog_dir, offline, capsys): + _write_entry( + catalog_dir, + "alpha-entry", + tier="official", + requires_hermes=">=0.19", + docs_url="https://example.com/docs", + platforms=["linux"], + capabilities={ + "provides_tools": ["cool_tool"], + "requires_env": ["ALPHA_KEY"], + }, + ) + plugins_cmd.cmd_info("alpha-entry") + out = capsys.readouterr().out + assert SHA_A in out + assert "official" in out + assert "cool_tool" in out + assert "ALPHA_KEY" in out + assert ">=0.19" in out + assert "hermes plugins install alpha-entry" in out + + def test_info_unknown_entry_exits(self, catalog_dir, offline, capsys): + with pytest.raises(SystemExit): + plugins_cmd.cmd_info("ghost-entry") + + def test_info_warns_when_removed(self, catalog_dir, offline, capsys): + _write_entry(catalog_dir, "alpha-entry") + _write_removed( + catalog_dir, + [{"name": "alpha-entry", "reason": "bad actor"}], + ) + plugins_cmd.cmd_info("alpha-entry") + out = capsys.readouterr().out + assert "REMOVED" in out + assert "bad actor" in out + + +# ── doctor ───────────────────────────────────────────────────────────────── + + +class TestDoctor: + @pytest.fixture(autouse=True) + def _no_runtime_scan(self, monkeypatch): + monkeypatch.setattr( + plugins_cmd, "_runtime_load_errors", lambda: {} + ) + + def test_doctor_table_lists_installed_plugin( + self, catalog_dir, offline, capsys + ): + _install_user_plugin( + "doc-plugin", + sidecar={ + "catalog_name": "doc-plugin", + "repo": "https://github.com/example/doc-plugin", + "sha": SHA_A, + "tier": "official", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + _write_entry(catalog_dir, "doc-plugin", sha=SHA_A, tier="official") + plugins_cmd.cmd_doctor() + out = capsys.readouterr().out + assert "doc-plugin" in out + assert "official" in out + + def test_doctor_detail_flags_pin_mismatch( + self, catalog_dir, offline, capsys + ): + _install_user_plugin( + "doc-plugin", + sidecar={ + "catalog_name": "doc-plugin", + "repo": "https://github.com/example/doc-plugin", + "sha": SHA_A, + "tier": "official", + "installed_at": "2026-01-01T00:00:00Z", + }, + ) + _write_entry(catalog_dir, "doc-plugin", sha=SHA_B) + plugins_cmd.cmd_doctor("doc-plugin") + out = capsys.readouterr().out + assert "doc-plugin" in out + assert "behind catalog pin" in out or "pin mismatch" in out + + def test_doctor_flags_removed_plugin(self, catalog_dir, offline, capsys): + _install_user_plugin("evil-plugin") + _write_removed( + catalog_dir, + [{"name": "evil-plugin", "reason": "exfiltrated env vars"}], + ) + plugins_cmd.cmd_doctor("evil-plugin") + out = capsys.readouterr().out + assert "REMOVED" in out + + def test_doctor_unknown_plugin_exits(self, catalog_dir, offline, capsys): + with pytest.raises(SystemExit): + plugins_cmd.cmd_doctor("no-such-plugin") + + +# ── argparse dispatch ────────────────────────────────────────────────────── + + +class TestDispatch: + def _dispatch(self, monkeypatch, action, **attrs): + recorded = {} + + def record(fn_name): + def _rec(*args, **kwargs): + recorded["fn"] = fn_name + recorded["args"] = args + recorded["kwargs"] = kwargs + + return _rec + + for fn in ( + "cmd_search", + "cmd_browse", + "cmd_info", + "cmd_validate", + "cmd_doctor", + "cmd_install", + ): + monkeypatch.setattr(plugins_cmd, fn, record(fn)) + ns = argparse.Namespace(plugins_action=action, **attrs) + plugins_cmd.plugins_command(ns) + return recorded + + def test_search_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "search", query="foo") + assert rec["fn"] == "cmd_search" + assert "foo" in rec["args"] or rec["kwargs"].get("query") == "foo" + + def test_browse_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "browse") + assert rec["fn"] == "cmd_browse" + + def test_info_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "info", name="foo") + assert rec["fn"] == "cmd_info" + + def test_validate_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "validate", path="/tmp/x", json=True) + assert rec["fn"] == "cmd_validate" + + def test_doctor_dispatch(self, monkeypatch): + rec = self._dispatch(monkeypatch, "doctor", name=None) + assert rec["fn"] == "cmd_doctor" + + def test_install_allow_removed_dispatch(self, monkeypatch): + rec = self._dispatch( + monkeypatch, + "install", + identifier="x", + force=False, + enable=False, + no_enable=True, + allow_removed=True, + ) + assert rec["fn"] == "cmd_install" + assert rec["kwargs"].get("allow_removed") is True + + def test_parser_wires_new_subcommands(self): + from hermes_cli.subcommands.plugins import build_plugins_parser + + parser = argparse.ArgumentParser() + sub = parser.add_subparsers(dest="command") + build_plugins_parser(sub, cmd_plugins=lambda args: None) + for argv in ( + ["plugins", "search", "foo"], + ["plugins", "browse"], + ["plugins", "info", "foo"], + ["plugins", "validate", "/tmp/x", "--json"], + ["plugins", "doctor"], + ["plugins", "install", "foo", "--allow-removed"], + ): + args = parser.parse_args(argv) + assert args.plugins_action == argv[1]