diff --git a/tests/hermes_cli/test_plugin_catalog.py b/tests/hermes_cli/test_plugin_catalog.py index 83ce653edd..19a3613515 100644 --- a/tests/hermes_cli/test_plugin_catalog.py +++ b/tests/hermes_cli/test_plugin_catalog.py @@ -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 /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 /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" diff --git a/tests/hermes_cli/test_plugin_packs.py b/tests/hermes_cli/test_plugin_packs.py index 5bf082b64a..c9c80cc73a 100644 --- a/tests/hermes_cli/test_plugin_packs.py +++ b/tests/hermes_cli/test_plugin_packs.py @@ -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(): diff --git a/tests/hermes_cli/test_plugin_validate.py b/tests/hermes_cli/test_plugin_validate.py index a728e9eed0..4398b517f8 100644 --- a/tests/hermes_cli/test_plugin_validate.py +++ b/tests/hermes_cli/test_plugin_validate.py @@ -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 diff --git a/tests/hermes_cli/test_plugins_cmd_catalog.py b/tests/hermes_cli/test_plugins_cmd_catalog.py index fdcd54ffa3..f42f1d0165 100644 --- a/tests/hermes_cli/test_plugins_cmd_catalog.py +++ b/tests/hermes_cli/test_plugins_cmd_catalog.py @@ -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" diff --git a/tests/hermes_cli/test_web_plugins_catalog.py b/tests/hermes_cli/test_web_plugins_catalog.py index 26cd3ef6ac..26c80aa440 100644 --- a/tests/hermes_cli/test_web_plugins_catalog.py +++ b/tests/hermes_cli/test_web_plugins_catalog.py @@ -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 diff --git a/tests/website/test_extract_plugins.py b/tests/website/test_extract_plugins.py index 6edb3c1242..c531ae0ff7 100644 --- a/tests/website/test_extract_plugins.py +++ b/tests/website/test_extract_plugins.py @@ -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):