test(plugins): trim catalog tests to invariants against the new shapes
538+570+296+216 lines of change-detectors → 4 files of contract tests: seed catalog valid, bad entries skipped, kill list name-or-repo, live fallback+union; real-git pinned install + sidecar + re-pin; kill list blocks CLI/dashboard/TUI with only the CLI bypass; dashboard merge via sidecar; probe get_config default; extractor live document.
This commit is contained in:
Teknium
@@ -1,538 +1,61 @@
|
||||
"""Tests for the Hermes plugin catalog (hermes_cli.plugin_catalog) and the
|
||||
catalog-driven install/manifest extensions in plugins_cmd.py / plugins.py."""
|
||||
"""Plugin catalog contracts (hermes_cli/plugin_catalog.py): the in-tree seed is valid, bad entries are
|
||||
skipped not raised, kill-list matching is name-or-repo, and the live catalog degrades to in-tree."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
import json
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from hermes_cli.plugin_catalog import (
|
||||
CATALOG_TIERS,
|
||||
PluginCatalogEntry,
|
||||
RemovedEntry,
|
||||
entry_capability_summary,
|
||||
find_removed,
|
||||
get_catalog_dir,
|
||||
get_catalog_entry,
|
||||
load_catalog,
|
||||
load_removed_list,
|
||||
search_catalog,
|
||||
)
|
||||
|
||||
|
||||
VALID_SHA = "38fe0fb53eff98d477f807432e965429e665ca33"
|
||||
|
||||
|
||||
# ── Helpers ────────────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
def _write_entry(catalog_dir: Path, name: str, **overrides) -> Path:
|
||||
"""Write a minimal valid catalog entry yaml, applying overrides."""
|
||||
data = {
|
||||
"name": name,
|
||||
"repo": f"https://github.com/example/{name}",
|
||||
"sha": VALID_SHA,
|
||||
"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
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def catalog_dir(tmp_path, monkeypatch):
|
||||
d = tmp_path / "catalog"
|
||||
d.mkdir()
|
||||
monkeypatch.setenv("HERMES_PLUGIN_CATALOG_DIR", str(d))
|
||||
return d
|
||||
|
||||
|
||||
# ── get_catalog_dir ────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestGetCatalogDir:
|
||||
def test_env_override_wins(self, catalog_dir):
|
||||
assert get_catalog_dir() == catalog_dir
|
||||
|
||||
def test_default_is_repo_plugin_catalog(self, monkeypatch):
|
||||
monkeypatch.delenv("HERMES_PLUGIN_CATALOG_DIR", raising=False)
|
||||
d = get_catalog_dir()
|
||||
assert d.name == "plugin-catalog"
|
||||
|
||||
|
||||
# ── load_catalog ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestLoadCatalog:
|
||||
def test_valid_entry_parses(self, catalog_dir):
|
||||
_write_entry(
|
||||
catalog_dir,
|
||||
"my-plugin",
|
||||
tier="official",
|
||||
requires_hermes=">=0.19",
|
||||
subdir="plugins/my-plugin",
|
||||
docs_url="https://example.com/docs",
|
||||
platforms=["linux"],
|
||||
capabilities={
|
||||
"provides_tools": ["my_tool"],
|
||||
"provides_hooks": ["on_start"],
|
||||
"provides_middleware": ["llm_request"],
|
||||
"requires_env": ["MY_API_KEY"],
|
||||
},
|
||||
)
|
||||
entries = load_catalog()
|
||||
assert len(entries) == 1
|
||||
e = entries[0]
|
||||
assert isinstance(e, PluginCatalogEntry)
|
||||
assert e.name == "my-plugin"
|
||||
assert e.repo == "https://github.com/example/my-plugin"
|
||||
assert e.sha == VALID_SHA
|
||||
assert e.tier == "official"
|
||||
assert e.requires_hermes == ">=0.19"
|
||||
assert e.subdir == "plugins/my-plugin"
|
||||
assert e.docs_url == "https://example.com/docs"
|
||||
assert e.platforms == ["linux"]
|
||||
assert e.capabilities.provides_tools == ["my_tool"]
|
||||
assert e.capabilities.provides_hooks == ["on_start"]
|
||||
assert e.capabilities.provides_middleware == ["llm_request"]
|
||||
assert e.capabilities.requires_env == ["MY_API_KEY"]
|
||||
|
||||
def test_tier_defaults_to_community(self, catalog_dir):
|
||||
_write_entry(catalog_dir, "no-tier")
|
||||
(entry,) = load_catalog()
|
||||
assert entry.tier == "community"
|
||||
assert entry.tier in CATALOG_TIERS
|
||||
|
||||
def test_bad_sha_rejected(self, catalog_dir, caplog):
|
||||
_write_entry(catalog_dir, "bad-sha", sha="main")
|
||||
_write_entry(catalog_dir, "short-sha", sha="38fe0fb")
|
||||
_write_entry(catalog_dir, "good", sha=VALID_SHA)
|
||||
with caplog.at_level("WARNING"):
|
||||
entries = load_catalog()
|
||||
assert [e.name for e in entries] == ["good"]
|
||||
|
||||
def test_bad_name_rejected(self, catalog_dir, caplog):
|
||||
_write_entry(catalog_dir, "BadName")
|
||||
_write_entry(catalog_dir, "has spaces")
|
||||
with caplog.at_level("WARNING"):
|
||||
entries = load_catalog()
|
||||
assert entries == []
|
||||
|
||||
def test_non_https_repo_rejected(self, catalog_dir, caplog):
|
||||
_write_entry(catalog_dir, "sshrepo", repo="git@github.com:x/y.git")
|
||||
with caplog.at_level("WARNING"):
|
||||
entries = load_catalog()
|
||||
assert entries == []
|
||||
|
||||
def test_invalid_tier_rejected(self, catalog_dir, caplog):
|
||||
_write_entry(catalog_dir, "weird-tier", tier="platinum")
|
||||
with caplog.at_level("WARNING"):
|
||||
entries = load_catalog()
|
||||
assert entries == []
|
||||
|
||||
def test_removed_yaml_is_not_an_entry(self, catalog_dir):
|
||||
_write_entry(catalog_dir, "real-entry")
|
||||
_write_removed(catalog_dir, [])
|
||||
entries = load_catalog()
|
||||
assert [e.name for e in entries] == ["real-entry"]
|
||||
|
||||
def test_unparseable_yaml_skipped_without_raising(self, catalog_dir, caplog):
|
||||
(catalog_dir / "broken.yaml").write_text(
|
||||
"name: [unclosed", encoding="utf-8"
|
||||
)
|
||||
_write_entry(catalog_dir, "ok-entry")
|
||||
with caplog.at_level("WARNING"):
|
||||
entries = load_catalog()
|
||||
assert [e.name for e in entries] == ["ok-entry"]
|
||||
|
||||
def test_missing_dir_returns_empty(self, tmp_path, monkeypatch):
|
||||
monkeypatch.setenv(
|
||||
"HERMES_PLUGIN_CATALOG_DIR", str(tmp_path / "does-not-exist")
|
||||
)
|
||||
assert load_catalog() == []
|
||||
|
||||
|
||||
# ── get_catalog_entry / search_catalog ─────────────────────────────────────
|
||||
|
||||
|
||||
class TestLookupAndSearch:
|
||||
def test_get_catalog_entry_by_name(self, catalog_dir):
|
||||
_write_entry(catalog_dir, "alpha")
|
||||
_write_entry(catalog_dir, "beta")
|
||||
entry = get_catalog_entry("beta")
|
||||
assert entry is not None and entry.name == "beta"
|
||||
assert get_catalog_entry("nope") is None
|
||||
|
||||
def test_search_matches_name_case_insensitive(self, catalog_dir):
|
||||
_write_entry(catalog_dir, "weather-tools")
|
||||
_write_entry(catalog_dir, "other")
|
||||
results = search_catalog("WEATHER")
|
||||
assert [e.name for e in results] == ["weather-tools"]
|
||||
|
||||
def test_search_matches_description(self, catalog_dir):
|
||||
_write_entry(catalog_dir, "abc", description="Fetches Stock Quotes.")
|
||||
results = search_catalog("stock")
|
||||
assert [e.name for e in results] == ["abc"]
|
||||
|
||||
def test_search_matches_declared_tools(self, catalog_dir):
|
||||
_write_entry(
|
||||
catalog_dir,
|
||||
"toolful",
|
||||
capabilities={"provides_tools": ["get_forecast"]},
|
||||
)
|
||||
_write_entry(catalog_dir, "toolless")
|
||||
results = search_catalog("Forecast")
|
||||
assert [e.name for e in results] == ["toolful"]
|
||||
|
||||
def test_empty_query_returns_all(self, catalog_dir):
|
||||
_write_entry(catalog_dir, "one")
|
||||
_write_entry(catalog_dir, "two")
|
||||
assert len(search_catalog("")) == 2
|
||||
|
||||
|
||||
# ── removed list ───────────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestRemovedList:
|
||||
def test_load_removed_list(self, catalog_dir):
|
||||
_write_removed(
|
||||
catalog_dir,
|
||||
[
|
||||
{
|
||||
"name": "evil-plugin",
|
||||
"repo": "https://github.com/evil/evil-plugin",
|
||||
"reason": "Exfiltrated env vars",
|
||||
"date": "2026-07-02",
|
||||
}
|
||||
],
|
||||
)
|
||||
removed = load_removed_list()
|
||||
assert len(removed) == 1
|
||||
r = removed[0]
|
||||
assert isinstance(r, RemovedEntry)
|
||||
assert r.name == "evil-plugin"
|
||||
assert r.reason == "Exfiltrated env vars"
|
||||
assert r.date == "2026-07-02"
|
||||
|
||||
def test_missing_removed_yaml_returns_empty(self, catalog_dir):
|
||||
assert load_removed_list() == []
|
||||
assert find_removed("anything") is None
|
||||
|
||||
def test_find_removed_by_name(self, catalog_dir):
|
||||
_write_removed(catalog_dir, [{"name": "evil-plugin", "reason": "bad"}])
|
||||
hit = find_removed("evil-plugin")
|
||||
assert hit is not None and hit.reason == "bad"
|
||||
|
||||
def test_find_removed_by_repo_url_with_and_without_git_suffix(
|
||||
self, catalog_dir
|
||||
):
|
||||
_write_removed(
|
||||
catalog_dir,
|
||||
[
|
||||
{
|
||||
"name": "evil-plugin",
|
||||
"repo": "https://github.com/evil/evil-plugin",
|
||||
"reason": "bad",
|
||||
}
|
||||
],
|
||||
)
|
||||
assert find_removed("https://github.com/evil/evil-plugin") is not None
|
||||
assert find_removed("https://github.com/evil/evil-plugin.git") is not None
|
||||
assert find_removed("https://github.com/good/fine.git") is None
|
||||
|
||||
|
||||
# ── entry_capability_summary ───────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestCapabilitySummary:
|
||||
def test_summary_contains_declared_capabilities(self):
|
||||
entry = PluginCatalogEntry(
|
||||
name="cap-plugin",
|
||||
repo="https://github.com/example/cap-plugin",
|
||||
sha=VALID_SHA,
|
||||
description="Does capable things.",
|
||||
maintainer="Example",
|
||||
)
|
||||
entry.capabilities.provides_tools = ["tool_a", "tool_b"]
|
||||
entry.capabilities.provides_hooks = ["session_start"]
|
||||
entry.capabilities.requires_env = ["CAP_API_KEY"]
|
||||
summary = entry_capability_summary(entry)
|
||||
assert "tool_a" in summary
|
||||
assert "tool_b" in summary
|
||||
assert "session_start" in summary
|
||||
assert "CAP_API_KEY" in summary
|
||||
|
||||
def test_summary_for_empty_capabilities_mentions_none(self):
|
||||
entry = PluginCatalogEntry(
|
||||
name="plain",
|
||||
repo="https://github.com/example/plain",
|
||||
sha=VALID_SHA,
|
||||
description="Plain.",
|
||||
maintainer="Example",
|
||||
)
|
||||
summary = entry_capability_summary(entry)
|
||||
assert summary # non-empty human text
|
||||
|
||||
|
||||
# ── shipped catalog seed ───────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestShippedCatalog:
|
||||
def test_shipped_catalog_entries_are_valid(self, monkeypatch):
|
||||
"""Every yaml shipped in <repo>/plugin-catalog must load cleanly."""
|
||||
monkeypatch.delenv("HERMES_PLUGIN_CATALOG_DIR", raising=False)
|
||||
shipped = get_catalog_dir()
|
||||
yaml_files = [
|
||||
p for p in shipped.glob("*.yaml") if p.name != "removed.yaml"
|
||||
]
|
||||
entries = load_catalog()
|
||||
assert len(entries) == len(yaml_files)
|
||||
# removed.yaml must exist and parse
|
||||
assert (shipped / "removed.yaml").exists()
|
||||
load_removed_list()
|
||||
|
||||
|
||||
# ── _version_satisfies ─────────────────────────────────────────────────────
|
||||
|
||||
|
||||
class TestVersionSatisfies:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _import(self):
|
||||
from hermes_cli.plugins import _version_satisfies
|
||||
|
||||
self.satisfies = _version_satisfies
|
||||
|
||||
def test_ge(self):
|
||||
assert self.satisfies(">=0.19", "0.19.0") is True
|
||||
assert self.satisfies(">=0.19", "0.20.1") is True
|
||||
assert self.satisfies(">=0.19", "0.18.2") is False
|
||||
|
||||
def test_gt_lt_le(self):
|
||||
assert self.satisfies(">0.19", "0.19.1") is True
|
||||
assert self.satisfies(">0.19", "0.19.0") is False
|
||||
assert self.satisfies("<1.0", "0.19.0") is True
|
||||
assert self.satisfies("<=0.19.0", "0.19.0") is True
|
||||
|
||||
def test_eq_ne(self):
|
||||
assert self.satisfies("==0.19.0", "0.19.0") is True
|
||||
assert self.satisfies("==0.19.0", "0.19.1") is False
|
||||
assert self.satisfies("!=0.19.0", "0.19.1") is True
|
||||
assert self.satisfies("!=0.19.0", "0.19.0") is False
|
||||
|
||||
def test_comma_separated_all_must_hold(self):
|
||||
assert self.satisfies(">=0.10, <1.0", "0.19.0") is True
|
||||
assert self.satisfies(">=0.10, <0.15", "0.19.0") is False
|
||||
|
||||
def test_bare_version_treated_as_ge(self):
|
||||
assert self.satisfies("0.10", "0.19.0") is True
|
||||
assert self.satisfies("999", "0.19.0") is False
|
||||
|
||||
def test_empty_spec_is_satisfied(self):
|
||||
assert self.satisfies("", "0.19.0") is True
|
||||
|
||||
def test_non_numeric_segments_fall_back_permissive(self):
|
||||
assert self.satisfies(">=abc.def", "0.19.0") is True
|
||||
assert self.satisfies(">=0.19", "unknown") is True
|
||||
|
||||
|
||||
# ── requires_hermes manifest gate ──────────────────────────────────────────
|
||||
|
||||
|
||||
def _make_plugin(base: Path, name: str, *, manifest_extra: dict | None = None,
|
||||
register_body: str = "pass", enable: bool = True) -> Path:
|
||||
"""Create a plugin dir under <HERMES_HOME>/plugins and opt it in."""
|
||||
plugin_dir = base / name
|
||||
plugin_dir.mkdir(parents=True, exist_ok=True)
|
||||
manifest = {"name": name, "version": "0.1.0", "description": name}
|
||||
if manifest_extra:
|
||||
manifest.update(manifest_extra)
|
||||
(plugin_dir / "plugin.yaml").write_text(yaml.safe_dump(manifest))
|
||||
(plugin_dir / "__init__.py").write_text(
|
||||
f"def register(ctx):\n {register_body}\n"
|
||||
)
|
||||
if enable:
|
||||
hermes_home = Path(os.environ["HERMES_HOME"])
|
||||
cfg_path = hermes_home / "config.yaml"
|
||||
cfg: dict = {}
|
||||
if cfg_path.exists():
|
||||
cfg = yaml.safe_load(cfg_path.read_text()) or {}
|
||||
cfg.setdefault("plugins", {}).setdefault("enabled", []).append(name)
|
||||
cfg_path.write_text(yaml.safe_dump(cfg))
|
||||
return plugin_dir
|
||||
|
||||
|
||||
class TestRequiresHermesGate:
|
||||
def test_unsatisfied_requires_hermes_skips_load(self, monkeypatch):
|
||||
from hermes_cli.plugins import PluginManager
|
||||
|
||||
hermes_home = Path(os.environ["HERMES_HOME"])
|
||||
plugins_dir = hermes_home / "plugins"
|
||||
_make_plugin(
|
||||
plugins_dir, "future_plugin",
|
||||
manifest_extra={"requires_hermes": ">=999.0"},
|
||||
)
|
||||
mgr = PluginManager()
|
||||
mgr.discover_and_load()
|
||||
loaded = mgr._plugins["future_plugin"]
|
||||
assert loaded.enabled is False
|
||||
assert loaded.error is not None
|
||||
assert "requires hermes" in loaded.error
|
||||
assert ">=999.0" in loaded.error
|
||||
assert loaded.module is None # register() never ran
|
||||
|
||||
def test_satisfied_requires_hermes_loads_normally(self, monkeypatch):
|
||||
from hermes_cli.plugins import PluginManager
|
||||
|
||||
hermes_home = Path(os.environ["HERMES_HOME"])
|
||||
plugins_dir = hermes_home / "plugins"
|
||||
_make_plugin(
|
||||
plugins_dir, "old_ok_plugin",
|
||||
manifest_extra={"requires_hermes": ">=0.1"},
|
||||
)
|
||||
mgr = PluginManager()
|
||||
mgr.discover_and_load()
|
||||
loaded = mgr._plugins["old_ok_plugin"]
|
||||
assert loaded.enabled is True
|
||||
assert loaded.error is None
|
||||
|
||||
def test_requires_hermes_parsed_onto_manifest(self):
|
||||
from hermes_cli.plugins import PluginManager
|
||||
|
||||
hermes_home = Path(os.environ["HERMES_HOME"])
|
||||
plugins_dir = hermes_home / "plugins"
|
||||
_make_plugin(
|
||||
plugins_dir, "spec_plugin",
|
||||
manifest_extra={"requires_hermes": ">=0.19"},
|
||||
enable=False,
|
||||
)
|
||||
mgr = PluginManager()
|
||||
mgr.discover_and_load()
|
||||
assert mgr._plugins["spec_plugin"].manifest.requires_hermes == ">=0.19"
|
||||
|
||||
|
||||
# ── _install_plugin_core: ref checkout + removed blocklist ────────────────
|
||||
|
||||
|
||||
def _make_git_repo(tmp_path: Path) -> tuple[Path, str, str]:
|
||||
"""Create a local git repo with two commits; return (path, sha1, sha2)."""
|
||||
repo = tmp_path / "src-repo"
|
||||
repo.mkdir()
|
||||
|
||||
def git(*args):
|
||||
subprocess.run(
|
||||
["git", *args], cwd=repo, check=True, capture_output=True, text=True,
|
||||
env={**os.environ,
|
||||
"GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
|
||||
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"},
|
||||
)
|
||||
|
||||
git("init", "-b", "main")
|
||||
(repo / "plugin.yaml").write_text(
|
||||
yaml.safe_dump({"name": "refplugin", "version": "1"})
|
||||
)
|
||||
(repo / "__init__.py").write_text("def register(ctx):\n pass\n")
|
||||
(repo / "marker.txt").write_text("first\n")
|
||||
git("add", "-A")
|
||||
git("commit", "-m", "first")
|
||||
sha1 = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"], cwd=repo, check=True,
|
||||
capture_output=True, text=True,
|
||||
).stdout.strip()
|
||||
(repo / "marker.txt").write_text("second\n")
|
||||
git("add", "-A")
|
||||
git("commit", "-m", "second")
|
||||
sha2 = subprocess.run(
|
||||
["git", "rev-parse", "HEAD"], cwd=repo, check=True,
|
||||
capture_output=True, text=True,
|
||||
).stdout.strip()
|
||||
return repo, sha1, sha2
|
||||
|
||||
|
||||
class TestInstallPluginCore:
|
||||
def test_ref_checkout_installs_pinned_commit(self, tmp_path, catalog_dir):
|
||||
from hermes_cli.plugins_cmd import _install_plugin_core
|
||||
|
||||
repo, sha1, _sha2 = _make_git_repo(tmp_path)
|
||||
target, manifest, name = _install_plugin_core(
|
||||
f"file://{repo}", force=False, ref=sha1
|
||||
)
|
||||
assert name == "refplugin"
|
||||
assert (target / "marker.txt").read_text() == "first\n"
|
||||
|
||||
def test_default_install_gets_head(self, tmp_path, catalog_dir):
|
||||
from hermes_cli.plugins_cmd import _install_plugin_core
|
||||
|
||||
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
|
||||
target, _manifest, _name = _install_plugin_core(
|
||||
f"file://{repo}", force=False
|
||||
)
|
||||
assert (target / "marker.txt").read_text() == "second\n"
|
||||
|
||||
def test_bad_ref_raises(self, tmp_path, catalog_dir):
|
||||
from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core
|
||||
|
||||
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
|
||||
with pytest.raises(PluginOperationError):
|
||||
_install_plugin_core(
|
||||
f"file://{repo}", force=False,
|
||||
ref="0000000000000000000000000000000000000000",
|
||||
)
|
||||
|
||||
def test_removed_repo_blocked(self, tmp_path, catalog_dir):
|
||||
from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core
|
||||
|
||||
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
|
||||
_write_removed(
|
||||
catalog_dir,
|
||||
[{
|
||||
"name": "refplugin",
|
||||
"repo": f"file://{repo}",
|
||||
"reason": "exfiltrated env vars",
|
||||
"date": "2026-07-02",
|
||||
}],
|
||||
)
|
||||
with pytest.raises(PluginOperationError, match="exfiltrated env vars"):
|
||||
_install_plugin_core(f"file://{repo}", force=False)
|
||||
|
||||
def test_removed_identifier_blocked_by_name(self, tmp_path, catalog_dir):
|
||||
from hermes_cli.plugins_cmd import PluginOperationError, _install_plugin_core
|
||||
|
||||
_write_removed(
|
||||
catalog_dir,
|
||||
[{"name": "evil-plugin", "reason": "malware", "date": "2026-01-01"}],
|
||||
)
|
||||
with pytest.raises(PluginOperationError, match="malware"):
|
||||
_install_plugin_core("evil-plugin", force=False)
|
||||
|
||||
def test_skip_removed_check_bypasses_block(self, tmp_path, catalog_dir):
|
||||
from hermes_cli.plugins_cmd import _install_plugin_core
|
||||
|
||||
repo, _sha1, _sha2 = _make_git_repo(tmp_path)
|
||||
_write_removed(
|
||||
catalog_dir,
|
||||
[{
|
||||
"name": "refplugin",
|
||||
"repo": f"file://{repo}",
|
||||
"reason": "bad",
|
||||
"date": "2026-07-02",
|
||||
}],
|
||||
)
|
||||
target, _manifest, name = _install_plugin_core(
|
||||
f"file://{repo}", force=False, skip_removed_check=True
|
||||
)
|
||||
assert name == "refplugin"
|
||||
assert target.exists()
|
||||
from hermes_cli import plugin_catalog as pc
|
||||
|
||||
SHA = "38fe0fb53eff98d477f807432e965429e665ca33"
|
||||
|
||||
|
||||
def _entry(name="good-plugin", **over):
|
||||
data = {"name": name, "repo": "https://github.com/owner/repo", "sha": SHA, "description": "d",
|
||||
"maintainer": "owner", "tier": "community", "capabilities": {"provides_tools": ["t1"]}}
|
||||
data.update(over)
|
||||
return data
|
||||
|
||||
|
||||
def test_shipped_catalog_entries_are_all_valid_and_pinned():
|
||||
"""Every file in plugin-catalog/ (minus removed.yaml) must parse — a dropped entry is a silent
|
||||
shipping regression the admission CI only catches on changed files."""
|
||||
root = pc.get_catalog_dir()
|
||||
files = [p for p in root.glob("*.yaml") if p.name != "removed.yaml"]
|
||||
entries = pc.load_catalog()
|
||||
assert len(entries) == len(files) >= 1
|
||||
assert all(pc._SHA_RE.match(e.sha) and e.repo.startswith("https://") for e in entries)
|
||||
assert all(e.install_identifier.startswith(e.repo) for e in entries)
|
||||
|
||||
|
||||
def test_invalid_entries_are_skipped_not_raised(tmp_path):
|
||||
(tmp_path / "a.yaml").write_text(yaml.safe_dump(_entry("ok")))
|
||||
(tmp_path / "b.yaml").write_text(yaml.safe_dump(_entry("short-sha", sha="abc123")))
|
||||
(tmp_path / "c.yaml").write_text(yaml.safe_dump(_entry("http-repo", repo="http://x/y")))
|
||||
(tmp_path / "d.yaml").write_text(yaml.safe_dump(_entry("Bad Name")))
|
||||
(tmp_path / "e.yaml").write_text("- not\n- a mapping\n")
|
||||
assert [e.name for e in pc.load_catalog(tmp_path)] == ["ok"]
|
||||
|
||||
|
||||
def test_find_removed_matches_name_or_normalized_repo(tmp_path):
|
||||
(tmp_path / "removed.yaml").write_text(yaml.safe_dump({"removed": [
|
||||
{"name": "evil", "repo": "https://github.com/x/evil.git", "reason": "malware", "date": "2026-01-01"}]}))
|
||||
assert pc.find_removed("evil", tmp_path).reason == "malware"
|
||||
assert pc.find_removed("https://github.com/x/EVIL/", tmp_path) is not None
|
||||
assert pc.find_removed("https://github.com/x/fine", tmp_path) is None
|
||||
|
||||
|
||||
def test_live_catalog_falls_back_to_in_tree_and_unions_removals(tmp_path, monkeypatch):
|
||||
"""Network failure → in-tree entries; a cached live doc contributes entries AND removals."""
|
||||
cache = tmp_path / "cache" / "plugin-catalog.json"
|
||||
monkeypatch.setattr(pc, "_live_cache_path", lambda: cache)
|
||||
monkeypatch.setattr(pc, "LIVE_CATALOG_URL", "http://127.0.0.1:9/nope") # unreachable
|
||||
assert [e.name for e in pc.load_catalog_live()] == [e.name for e in pc.load_catalog()]
|
||||
|
||||
cache.parent.mkdir(parents=True)
|
||||
cache.write_text(json.dumps({"entries": [_entry("live-only")],
|
||||
"removed": [{"name": "pulled-live", "reason": "cve"}]}))
|
||||
assert [e.name for e in pc.load_catalog_live()] == ["live-only"]
|
||||
assert pc.find_removed("pulled-live").reason == "cve"
|
||||
|
||||
@@ -186,17 +186,10 @@ def test_load_pack_missing_file_errors():
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _fake_catalog_entry(name):
|
||||
return SimpleNamespace(
|
||||
name=name,
|
||||
repo="https://github.com/idx-owner/idx-repo",
|
||||
subdir="",
|
||||
capabilities=SimpleNamespace(
|
||||
provides_tools=["tools"],
|
||||
provides_hooks=[],
|
||||
provides_middleware=[],
|
||||
requires_env=[],
|
||||
),
|
||||
)
|
||||
from hermes_cli.plugin_catalog import CatalogCapabilities, PluginCatalogEntry
|
||||
return PluginCatalogEntry(
|
||||
name=name, repo="https://github.com/idx-owner/idx-repo", sha=SHA_B, description="d", maintainer="idx-owner",
|
||||
capabilities=CatalogCapabilities(provides_tools=["tools"]))
|
||||
|
||||
|
||||
def test_resolve_pack_plugins_uses_catalog_for_bare_names():
|
||||
|
||||
@@ -6,13 +6,10 @@ 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
|
||||
|
||||
|
||||
@@ -36,86 +33,6 @@ BASE_MANIFEST = {
|
||||
}
|
||||
|
||||
|
||||
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_schema_fails(self, tmp_path):
|
||||
manifest = dict(
|
||||
BASE_MANIFEST, config_schema={"endpoint": {"type": "no-such-type"}}
|
||||
)
|
||||
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_schema_passes(self, tmp_path):
|
||||
manifest = dict(
|
||||
BASE_MANIFEST,
|
||||
config_schema={
|
||||
"endpoint": {"type": "str", "description": "Endpoint?"},
|
||||
"retries": {"type": "int", "default": 3},
|
||||
},
|
||||
)
|
||||
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 = (
|
||||
@@ -182,35 +99,16 @@ class TestCapabilityProbe:
|
||||
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
|
||||
def test_probe_context_returns_get_config_defaults(self, tmp_path):
|
||||
"""Real PluginContext.get_config yields the default when nothing is configured; the probe must
|
||||
too, or every plugin doing ``int(ctx.get_config("timeout", 180))`` fails admission."""
|
||||
d = _make_plugin(
|
||||
tmp_path,
|
||||
manifest={**BASE_MANIFEST, "provides_tools": ["t"]},
|
||||
init_py=(
|
||||
"def register(ctx):\n"
|
||||
" int(ctx.get_config('timeout_seconds', 180))\n"
|
||||
" ctx.register_tool('t', schema={}, handler=lambda **kw: None)\n"),
|
||||
)
|
||||
report = validate_plugin_dir(d)
|
||||
assert report.ok, report.failures
|
||||
|
||||
@@ -1,570 +1,98 @@
|
||||
"""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.
|
||||
"""
|
||||
"""Catalog-aware ``hermes plugins`` surface (hermes_cli/plugins_cmd_catalog.py): a bare catalog name installs
|
||||
the PINNED sha and records provenance; the kill list blocks every install path (CLI needs an explicit
|
||||
bypass, dashboard/TUI have none); ``update`` re-pins instead of pulling. Real git, file:// repos."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import types
|
||||
import os
|
||||
import shutil
|
||||
import subprocess as sp
|
||||
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,
|
||||
scan_decision_cb=None):
|
||||
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 matched" 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 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_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", term="foo")
|
||||
assert rec["fn"] == "cmd_search"
|
||||
assert "foo" in rec["args"] or rec["kwargs"].get("term") == "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):
|
||||
# `doctor` is owned by the runtime-contract dev doctor on main.
|
||||
recorded = {}
|
||||
monkeypatch.setattr(
|
||||
plugins_cmd, "cmd_plugin_doctor",
|
||||
lambda target, *, ci=False: recorded.setdefault("target", target),
|
||||
)
|
||||
ns = argparse.Namespace(plugins_action="doctor", target=".", ci=False)
|
||||
plugins_cmd.plugins_command(ns)
|
||||
assert recorded["target"] == "."
|
||||
|
||||
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]
|
||||
from hermes_cli import plugin_catalog as pc_cat
|
||||
from hermes_cli import plugins_cmd as pc
|
||||
from hermes_cli import plugins_cmd_catalog as cat
|
||||
|
||||
pytestmark = pytest.mark.skipif(shutil.which("git") is None, reason="git not available")
|
||||
|
||||
_GIT_ENV = {**os.environ, "GIT_AUTHOR_NAME": "t", "GIT_AUTHOR_EMAIL": "t@t",
|
||||
"GIT_COMMITTER_NAME": "t", "GIT_COMMITTER_EMAIL": "t@t"}
|
||||
|
||||
|
||||
def _commit(repo: Path, msg: str) -> str:
|
||||
sp.run(["git", "add", "-A"], cwd=repo, check=True, env=_GIT_ENV)
|
||||
sp.run(["git", "commit", "-q", "-m", msg], cwd=repo, check=True, env=_GIT_ENV)
|
||||
return sp.run(["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True).stdout.strip()
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def world(tmp_path, monkeypatch):
|
||||
"""A file:// plugin repo with two commits, a catalog pinned to the FIRST, an isolated plugins dir."""
|
||||
repo = tmp_path / "repo"
|
||||
repo.mkdir()
|
||||
(repo / "plugin.yaml").write_text("name: cat-plugin\nversion: 1.0.0\ndescription: d\n")
|
||||
(repo / "__init__.py").write_text("def register(ctx):\n pass\n")
|
||||
sp.run(["git", "init", "-q"], cwd=repo, check=True, env=_GIT_ENV)
|
||||
sha1 = _commit(repo, "v1")
|
||||
(repo / "__init__.py").write_text("def register(ctx):\n pass # v2\n")
|
||||
sha2 = _commit(repo, "v2")
|
||||
|
||||
plugins_dir = tmp_path / "plugins"
|
||||
plugins_dir.mkdir()
|
||||
monkeypatch.setattr(pc, "_plugins_dir", lambda: plugins_dir)
|
||||
monkeypatch.setattr(pc, "_scan_on_install_enabled", lambda: False)
|
||||
monkeypatch.setattr(pc, "_console", lambda: type("C", (), {"print": lambda *a, **k: None})())
|
||||
|
||||
# Catalog: one entry pinned to sha1, mutable via state["pin"]; kill list via state["removed"]. The
|
||||
# real loader is https-only, so the fixture entry is built directly (file:// repo).
|
||||
state = {"pin": sha1, "removed": []}
|
||||
|
||||
def _entries():
|
||||
return [pc_cat.PluginCatalogEntry(name="cat-plugin", repo=repo.as_uri(), sha=state["pin"],
|
||||
description="d", maintainer="t")]
|
||||
|
||||
monkeypatch.setattr(pc_cat, "load_catalog", lambda catalog_dir=None: _entries())
|
||||
monkeypatch.setattr(pc_cat, "fetch_live_catalog", lambda **_: None) # in-tree only, no network
|
||||
monkeypatch.setattr(pc_cat, "load_removed_list", lambda catalog_dir=None: list(state["removed"]))
|
||||
return {"repo": repo, "sha1": sha1, "sha2": sha2, "plugins_dir": plugins_dir, "state": state}
|
||||
|
||||
|
||||
def _head(path: Path) -> str:
|
||||
return sp.run(["git", "rev-parse", "HEAD"], cwd=path, capture_output=True, text=True).stdout.strip()
|
||||
|
||||
|
||||
def test_catalog_name_installs_pinned_sha_with_sidecar_then_update_repins(world, monkeypatch):
|
||||
entry = pc_cat.get_live_catalog_entry("cat-plugin")
|
||||
assert entry is not None
|
||||
target, _m, name = cat.install_catalog_entry(entry, force=False)
|
||||
assert name == "cat-plugin"
|
||||
assert _head(target) == world["sha1"] != world["sha2"] # pinned, not HEAD
|
||||
sidecar = json.loads((target / cat.CATALOG_SIDECAR).read_text())
|
||||
assert (sidecar["catalog_name"], sidecar["sha"]) == ("cat-plugin", world["sha1"])
|
||||
assert cat.catalog_annotation(target) == f"catalog:community@{world['sha1'][:8]}"
|
||||
|
||||
# Dashboard update on a catalog install = re-pin. Pin unchanged → no-op.
|
||||
assert pc.dashboard_update_user_plugin("cat-plugin") == {
|
||||
"ok": True, "name": "cat-plugin", "sha": world["sha1"], "unchanged": True}
|
||||
# Bump the catalog pin → the checkout moves to exactly that sha.
|
||||
world["state"]["pin"] = world["sha2"]
|
||||
assert pc.dashboard_update_user_plugin("cat-plugin")["unchanged"] is False
|
||||
assert _head(world["plugins_dir"] / "cat-plugin") == world["sha2"]
|
||||
|
||||
|
||||
def test_kill_list_blocks_cli_dashboard_and_tui_paths(world, monkeypatch):
|
||||
world["state"]["removed"].append(
|
||||
pc_cat.RemovedEntry(name="cat-plugin", repo=world["repo"].as_uri(), reason="malware"))
|
||||
# Dashboard/TUI: catalog name AND raw repo URL both refused, no bypass parameter exists.
|
||||
assert "malware" in pc.dashboard_install_plugin("", force=False, enable=False, catalog_name="cat-plugin")["error"]
|
||||
assert "malware" in pc.dashboard_install_plugin(world["repo"].as_uri(), force=False, enable=False)["error"]
|
||||
assert not (world["plugins_dir"] / "cat-plugin").exists()
|
||||
# CLI: refused by default, `--allow-removed` installs anyway.
|
||||
with pytest.raises(SystemExit):
|
||||
pc.cmd_install("cat-plugin", enable=False)
|
||||
pc.cmd_install("cat-plugin", enable=False, allow_removed=True)
|
||||
assert (world["plugins_dir"] / "cat-plugin" / cat.CATALOG_SIDECAR).exists()
|
||||
assert cat.removed_annotation("cat-plugin", world["plugins_dir"] / "cat-plugin") == "malware"
|
||||
|
||||
@@ -1,296 +1,77 @@
|
||||
"""Tests for the dashboard plugin-catalog surface in hermes_cli.web_server.
|
||||
|
||||
Covers:
|
||||
- GET /api/dashboard/plugins/catalog — entry serialization, installed-state
|
||||
merge (via the ``.hermes-catalog.json`` sidecar), removed list exposure.
|
||||
- POST /api/dashboard/agent-plugins/install — removed-blocklist refusal for
|
||||
raw identifiers AND catalog names, catalog_name resolution to a pinned-ref
|
||||
install, sidecar write.
|
||||
- /api/dashboard/plugins/hub — ``removed_reason`` annotation on rows.
|
||||
"""
|
||||
"""Dashboard plugin-catalog surface: GET /api/dashboard/plugins/catalog merges installed state (via the
|
||||
``.hermes-catalog.json`` sidecar) and the install endpoint has NO kill-list bypass."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from hermes_cli import plugin_catalog as pc_cat
|
||||
|
||||
VALID_SHA = "38fe0fb53eff98d477f807432e965429e665ca33"
|
||||
OTHER_SHA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
||||
|
||||
|
||||
def _write_entry(catalog_dir: Path, name: str, **overrides) -> dict:
|
||||
data = {
|
||||
"name": name,
|
||||
"repo": f"https://github.com/example/{name}",
|
||||
"sha": VALID_SHA,
|
||||
"description": f"Test entry {name}.",
|
||||
"maintainer": "Example",
|
||||
"tier": "official",
|
||||
"docs_url": f"https://example.com/docs/{name}",
|
||||
"capabilities": {
|
||||
"provides_tools": ["tool_a"],
|
||||
"provides_hooks": ["hook_b"],
|
||||
"provides_middleware": [],
|
||||
"requires_env": ["EXAMPLE_API_KEY"],
|
||||
},
|
||||
}
|
||||
data.update(overrides)
|
||||
catalog_dir.mkdir(parents=True, exist_ok=True)
|
||||
(catalog_dir / f"{name}.yaml").write_text(
|
||||
yaml.safe_dump(data), encoding="utf-8"
|
||||
)
|
||||
return data
|
||||
|
||||
|
||||
def _write_removed(catalog_dir: Path, removed: list) -> None:
|
||||
catalog_dir.mkdir(parents=True, exist_ok=True)
|
||||
(catalog_dir / "removed.yaml").write_text(
|
||||
yaml.safe_dump({"removed": removed}), encoding="utf-8"
|
||||
)
|
||||
|
||||
|
||||
def _make_installed_plugin(name: str, sidecar: dict | None = None) -> Path:
|
||||
"""Drop a minimal plugin dir under the isolated HERMES_HOME."""
|
||||
from hermes_constants import get_hermes_home
|
||||
|
||||
plugin_dir = get_hermes_home() / "plugins" / name
|
||||
plugin_dir.mkdir(parents=True, exist_ok=True)
|
||||
(plugin_dir / "plugin.yaml").write_text(
|
||||
yaml.safe_dump({"name": name, "version": "1.0", "description": "x"}),
|
||||
encoding="utf-8",
|
||||
)
|
||||
if sidecar is not None:
|
||||
(plugin_dir / ".hermes-catalog.json").write_text(
|
||||
json.dumps(sidecar), encoding="utf-8"
|
||||
)
|
||||
return plugin_dir
|
||||
|
||||
|
||||
class TestDashboardPluginCatalog:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _setup(self, monkeypatch, tmp_path, _isolate_hermes_home):
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
except ImportError:
|
||||
pytest.skip("fastapi/starlette not installed")
|
||||
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
|
||||
monkeypatch.setattr(
|
||||
hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db"
|
||||
)
|
||||
|
||||
self.catalog_dir = tmp_path / "catalog"
|
||||
self.catalog_dir.mkdir()
|
||||
monkeypatch.setenv("HERMES_PLUGIN_CATALOG_DIR", str(self.catalog_dir))
|
||||
|
||||
self.client = TestClient(app)
|
||||
self.client.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
|
||||
# ── GET /api/dashboard/plugins/catalog ──────────────────────────────
|
||||
|
||||
def test_catalog_endpoint_requires_token(self):
|
||||
@pytest.fixture
|
||||
def client(monkeypatch, tmp_path, _isolate_hermes_home):
|
||||
try:
|
||||
from starlette.testclient import TestClient
|
||||
from hermes_cli.web_server import app
|
||||
except ImportError:
|
||||
pytest.skip("fastapi/starlette not installed")
|
||||
import hermes_state
|
||||
from hermes_constants import get_hermes_home
|
||||
from hermes_cli.web_server import app, _SESSION_HEADER_NAME, _SESSION_TOKEN
|
||||
|
||||
unauth = TestClient(app)
|
||||
resp = unauth.get("/api/dashboard/plugins/catalog")
|
||||
assert resp.status_code == 401
|
||||
monkeypatch.setattr(hermes_state, "DEFAULT_DB_PATH", get_hermes_home() / "state.db")
|
||||
catalog_dir = tmp_path / "catalog"
|
||||
catalog_dir.mkdir()
|
||||
(catalog_dir / "alpha-plugin.yaml").write_text(yaml.safe_dump({
|
||||
"name": "alpha-plugin", "repo": "https://github.com/example/alpha-plugin", "sha": VALID_SHA,
|
||||
"description": "d", "maintainer": "Example", "tier": "official",
|
||||
"capabilities": {"provides_tools": ["tool_a"], "requires_env": ["EXAMPLE_API_KEY"]}}))
|
||||
(catalog_dir / "removed.yaml").write_text(yaml.safe_dump({"removed": [
|
||||
{"name": "bad-plugin", "repo": "https://github.com/evil/bad-plugin", "reason": "exfiltrated env vars"}]}))
|
||||
monkeypatch.setattr(pc_cat, "get_catalog_dir", lambda: catalog_dir)
|
||||
monkeypatch.setattr(pc_cat, "fetch_live_catalog", lambda **_: None)
|
||||
|
||||
def test_catalog_endpoint_shape(self):
|
||||
_write_entry(self.catalog_dir, "alpha-plugin")
|
||||
_write_removed(
|
||||
self.catalog_dir,
|
||||
[{"name": "bad-plugin", "repo": "https://github.com/evil/bad-plugin",
|
||||
"reason": "exfiltrated env vars", "date": "2026-07-02"}],
|
||||
)
|
||||
c = TestClient(app)
|
||||
c.headers[_SESSION_HEADER_NAME] = _SESSION_TOKEN
|
||||
return c
|
||||
|
||||
resp = self.client.get("/api/dashboard/plugins/catalog")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
|
||||
assert "generated_at" in data
|
||||
assert isinstance(data["entries"], list) and len(data["entries"]) == 1
|
||||
entry = data["entries"][0]
|
||||
assert entry["name"] == "alpha-plugin"
|
||||
assert entry["repo"] == "https://github.com/example/alpha-plugin"
|
||||
assert entry["sha"] == VALID_SHA
|
||||
assert entry["sha_short"] == VALID_SHA[:7]
|
||||
assert entry["tier"] == "official"
|
||||
assert entry["maintainer"] == "Example"
|
||||
assert entry["docs_url"] == "https://example.com/docs/alpha-plugin"
|
||||
assert entry["capabilities"]["provides_tools"] == ["tool_a"]
|
||||
assert entry["capabilities"]["requires_env"] == ["EXAMPLE_API_KEY"]
|
||||
assert "tool_a" in entry["capability_summary"]
|
||||
# Not installed → degraded install state.
|
||||
assert entry["installed"] is False
|
||||
assert entry["installed_sha"] is None
|
||||
assert entry["update_available"] is False
|
||||
assert entry["runtime_status"] is None
|
||||
def _install(name: str, sidecar: dict | None):
|
||||
from hermes_constants import get_hermes_home
|
||||
d = get_hermes_home() / "plugins" / name
|
||||
d.mkdir(parents=True)
|
||||
(d / "plugin.yaml").write_text(yaml.safe_dump({"name": name, "version": "1.0", "description": "x"}))
|
||||
if sidecar:
|
||||
(d / ".hermes-catalog.json").write_text(json.dumps(sidecar))
|
||||
|
||||
assert len(data["removed"]) == 1
|
||||
removed = data["removed"][0]
|
||||
assert removed["name"] == "bad-plugin"
|
||||
assert removed["reason"] == "exfiltrated env vars"
|
||||
|
||||
def test_catalog_installed_state_merge_with_sidecar(self):
|
||||
_write_entry(self.catalog_dir, "alpha-plugin")
|
||||
_make_installed_plugin(
|
||||
"alpha-plugin",
|
||||
sidecar={
|
||||
"catalog_name": "alpha-plugin",
|
||||
"repo": "https://github.com/example/alpha-plugin",
|
||||
"sha": OTHER_SHA,
|
||||
"installed_at": "2026-07-01T00:00:00Z",
|
||||
"tier": "official",
|
||||
},
|
||||
)
|
||||
def test_catalog_endpoint_merges_installed_state_from_sidecar(client):
|
||||
from starlette.testclient import TestClient
|
||||
from hermes_cli.web_server import app
|
||||
assert TestClient(app).get("/api/dashboard/plugins/catalog").status_code == 401
|
||||
|
||||
resp = self.client.get("/api/dashboard/plugins/catalog")
|
||||
assert resp.status_code == 200
|
||||
entry = resp.json()["entries"][0]
|
||||
assert entry["installed"] is True
|
||||
assert entry["installed_sha"] == OTHER_SHA
|
||||
assert entry["update_available"] is True
|
||||
assert entry["runtime_status"] == "inactive"
|
||||
# Manifest name differs from the catalog name (the common case): matched through the sidecar.
|
||||
_install("alpha", {"catalog_name": "alpha-plugin", "sha": OTHER_SHA, "tier": "official"})
|
||||
data = client.get("/api/dashboard/plugins/catalog").json()
|
||||
[entry] = data["entries"]
|
||||
assert (entry["name"], entry["sha_short"], entry["capabilities"]["provides_tools"]) == ("alpha-plugin", VALID_SHA[:7], ["tool_a"])
|
||||
assert "tool_a" in entry["capability_summary"]
|
||||
assert (entry["installed"], entry["installed_sha"], entry["update_available"]) == (True, OTHER_SHA, True)
|
||||
assert entry["runtime_status"] == "inactive"
|
||||
assert data["removed"][0]["reason"] == "exfiltrated env vars"
|
||||
|
||||
def test_catalog_installed_no_sidecar_degrades_to_null_sha(self):
|
||||
_write_entry(self.catalog_dir, "alpha-plugin")
|
||||
_make_installed_plugin("alpha-plugin", sidecar=None)
|
||||
|
||||
resp = self.client.get("/api/dashboard/plugins/catalog")
|
||||
entry = resp.json()["entries"][0]
|
||||
assert entry["installed"] is True
|
||||
assert entry["installed_sha"] is None
|
||||
assert entry["update_available"] is False
|
||||
|
||||
def test_catalog_installed_same_sha_no_update(self):
|
||||
_write_entry(self.catalog_dir, "alpha-plugin")
|
||||
_make_installed_plugin(
|
||||
"alpha-plugin",
|
||||
sidecar={
|
||||
"catalog_name": "alpha-plugin",
|
||||
"repo": "https://github.com/example/alpha-plugin",
|
||||
"sha": VALID_SHA,
|
||||
"installed_at": "2026-07-01T00:00:00Z",
|
||||
"tier": "official",
|
||||
},
|
||||
)
|
||||
|
||||
entry = self.client.get("/api/dashboard/plugins/catalog").json()["entries"][0]
|
||||
assert entry["installed"] is True
|
||||
assert entry["installed_sha"] == VALID_SHA
|
||||
assert entry["update_available"] is False
|
||||
|
||||
# ── POST /api/dashboard/agent-plugins/install ───────────────────────
|
||||
|
||||
def test_install_refuses_removed_raw_identifier(self):
|
||||
_write_removed(
|
||||
self.catalog_dir,
|
||||
[{"name": "bad-plugin", "repo": "https://github.com/evil/bad-plugin",
|
||||
"reason": "exfiltrated env vars", "date": "2026-07-02"}],
|
||||
)
|
||||
resp = self.client.post(
|
||||
"/api/dashboard/agent-plugins/install",
|
||||
json={"identifier": "https://github.com/evil/bad-plugin"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "exfiltrated env vars" in resp.json()["detail"]
|
||||
|
||||
def test_install_refuses_removed_catalog_name(self):
|
||||
_write_removed(
|
||||
self.catalog_dir,
|
||||
[{"name": "bad-plugin", "reason": "policy violation",
|
||||
"date": "2026-07-02"}],
|
||||
)
|
||||
resp = self.client.post(
|
||||
"/api/dashboard/agent-plugins/install",
|
||||
json={"identifier": "", "catalog_name": "bad-plugin"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "policy violation" in resp.json()["detail"]
|
||||
|
||||
def test_install_unknown_catalog_name_is_400(self):
|
||||
resp = self.client.post(
|
||||
"/api/dashboard/agent-plugins/install",
|
||||
json={"identifier": "", "catalog_name": "does-not-exist"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "does-not-exist" in resp.json()["detail"]
|
||||
|
||||
def test_install_missing_identifier_and_catalog_name_is_400(self):
|
||||
resp = self.client.post(
|
||||
"/api/dashboard/agent-plugins/install",
|
||||
json={"identifier": ""},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_catalog_name_install_resolves_pinned_ref_and_writes_sidecar(
|
||||
self, monkeypatch, tmp_path
|
||||
):
|
||||
from hermes_constants import get_hermes_home
|
||||
import hermes_cli.plugins_cmd as plugins_cmd
|
||||
|
||||
_write_entry(self.catalog_dir, "alpha-plugin")
|
||||
|
||||
captured = {}
|
||||
|
||||
def fake_core(identifier, *, force, ref=None, skip_removed_check=False):
|
||||
captured["identifier"] = identifier
|
||||
captured["ref"] = ref
|
||||
target = get_hermes_home() / "plugins" / "alpha-plugin"
|
||||
target.mkdir(parents=True, exist_ok=True)
|
||||
(target / "plugin.yaml").write_text(
|
||||
yaml.safe_dump({"name": "alpha-plugin"}), encoding="utf-8"
|
||||
)
|
||||
return target, {"name": "alpha-plugin"}, "alpha-plugin"
|
||||
|
||||
monkeypatch.setattr(plugins_cmd, "_install_plugin_core", fake_core)
|
||||
|
||||
resp = self.client.post(
|
||||
"/api/dashboard/agent-plugins/install",
|
||||
json={"identifier": "", "catalog_name": "alpha-plugin",
|
||||
"enable": False},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["ok"] is True
|
||||
assert body["plugin_name"] == "alpha-plugin"
|
||||
|
||||
assert captured["ref"] == VALID_SHA
|
||||
assert captured["identifier"].startswith(
|
||||
"https://github.com/example/alpha-plugin"
|
||||
)
|
||||
|
||||
sidecar_path = (
|
||||
get_hermes_home() / "plugins" / "alpha-plugin" / ".hermes-catalog.json"
|
||||
)
|
||||
assert sidecar_path.is_file()
|
||||
sidecar = json.loads(sidecar_path.read_text(encoding="utf-8"))
|
||||
assert sidecar["catalog_name"] == "alpha-plugin"
|
||||
assert sidecar["repo"] == "https://github.com/example/alpha-plugin"
|
||||
assert sidecar["sha"] == VALID_SHA
|
||||
assert sidecar["tier"] == "official"
|
||||
assert sidecar["installed_at"]
|
||||
|
||||
# ── /api/dashboard/plugins/hub removed_reason annotation ─────────────
|
||||
|
||||
def test_hub_rows_annotated_with_removed_reason(self):
|
||||
_write_removed(
|
||||
self.catalog_dir,
|
||||
[{"name": "bad-plugin", "reason": "supply chain incident",
|
||||
"date": "2026-07-02"}],
|
||||
)
|
||||
_make_installed_plugin("bad-plugin")
|
||||
_make_installed_plugin("good-plugin")
|
||||
|
||||
resp = self.client.get("/api/dashboard/plugins/hub")
|
||||
assert resp.status_code == 200
|
||||
rows = {r["name"]: r for r in resp.json()["plugins"]}
|
||||
assert rows["bad-plugin"]["removed_reason"] == "supply chain incident"
|
||||
assert rows["good-plugin"]["removed_reason"] is None
|
||||
def test_install_endpoint_refuses_removed_plugins_with_no_bypass(client, monkeypatch):
|
||||
from hermes_cli import plugins_cmd
|
||||
monkeypatch.setattr(plugins_cmd, "_install_plugin_core", lambda *a, **k: pytest.fail("kill-listed install ran"))
|
||||
for body in ({"identifier": "https://github.com/evil/bad-plugin.git"},
|
||||
{"identifier": "", "catalog_name": "bad-plugin"}, {"identifier": "evil/bad-plugin"}):
|
||||
resp = client.post("/api/dashboard/agent-plugins/install", json=body)
|
||||
assert resp.status_code == 400, body
|
||||
assert "removed" in resp.json()["detail"] or "not in the Hermes plugin catalog" in resp.json()["detail"]
|
||||
assert client.post("/api/dashboard/agent-plugins/install", json={"identifier": ""}).status_code == 400
|
||||
|
||||
@@ -157,6 +157,11 @@ def test_main_writes_catalog_and_meta(mod, tmp_path):
|
||||
assert meta["byTier"] == {"official": 1, "community": 1}
|
||||
assert meta["removedCount"] == 1
|
||||
assert meta["generatedAt"]
|
||||
# The live-refresh document consumed by installed clients: loader-schema entries + the kill list.
|
||||
from hermes_cli.plugin_catalog import entry_from_mapping
|
||||
live = json.loads((out_dir / "plugin-catalog.json").read_text(encoding="utf-8"))
|
||||
assert [entry_from_mapping(raw, "live").name for raw in live["entries"]] == ["alpha", "beta"]
|
||||
assert live["removed"] == [{"name": "gone"}]
|
||||
|
||||
|
||||
def test_missing_catalog_dir_degrades_to_empty_outputs_exit_zero(mod, tmp_path):
|
||||
|
||||
Reference in New Issue
Block a user