Merge remote-tracking branch 'origin/main' into sid/types-and-lints

# Conflicts:
#	gateway/run.py
#	tools/delegate_tool.py
This commit is contained in:
alt-glitch 2026-04-22 14:27:37 +05:30
commit 847ffca715
171 changed files with 15125 additions and 1675 deletions

View file

@ -71,7 +71,11 @@ class TestProviderRegistry:
def test_kimi_env_vars(self):
pconfig = PROVIDER_REGISTRY["kimi-coding"]
assert pconfig.api_key_env_vars == ("KIMI_API_KEY",)
# KIMI_API_KEY is the primary env var; KIMI_CODING_API_KEY is a
# secondary fallback for Kimi Code sk-kimi- keys so users don't
# have to overload the same variable.
assert "KIMI_API_KEY" in pconfig.api_key_env_vars
assert "KIMI_CODING_API_KEY" in pconfig.api_key_env_vars
assert pconfig.base_url_env_var == "KIMI_BASE_URL"
def test_minimax_env_vars(self):

View file

@ -0,0 +1,90 @@
"""Regression test: `@folder:` completion must only surface directories and
`@file:` must only surface regular files.
Reported during TUI v2 blitz testing: typing `@folder:` showed .dockerignore,
.env, .gitignore, etc. alongside the actual directories because the path-
completion branch yielded every entry regardless of the explicit prefix, and
auto-switched the completion kind based on `is_dir`. That defeated the user's
explicit choice and rendered the `@folder:` / `@file:` prefixes useless for
filtering.
"""
from __future__ import annotations
from pathlib import Path
from typing import Iterable
from hermes_cli.commands import SlashCommandCompleter
def _run(tmp_path: Path, word: str) -> list[tuple[str, str]]:
(tmp_path / "readme.md").write_text("x")
(tmp_path / ".env").write_text("x")
(tmp_path / "src").mkdir()
(tmp_path / "docs").mkdir()
completer = SlashCommandCompleter.__new__(SlashCommandCompleter)
completions: Iterable = completer._context_completions(word)
return [(c.text, c.display_meta) for c in completions if c.text.startswith(("@file:", "@folder:"))]
def test_at_folder_only_yields_directories(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
texts = [t for t, _ in _run(tmp_path, "@folder:")]
assert all(t.startswith("@folder:") for t in texts), texts
assert any(t == "@folder:src/" for t in texts)
assert any(t == "@folder:docs/" for t in texts)
assert not any(t == "@folder:readme.md" for t in texts)
assert not any(t == "@folder:.env" for t in texts)
def test_at_file_only_yields_files(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
texts = [t for t, _ in _run(tmp_path, "@file:")]
assert all(t.startswith("@file:") for t in texts), texts
assert any(t == "@file:readme.md" for t in texts)
assert any(t == "@file:.env" for t in texts)
assert not any(t == "@file:src/" for t in texts)
assert not any(t == "@file:docs/" for t in texts)
def test_at_folder_preserves_prefix_on_empty_match(tmp_path, monkeypatch):
"""User typed `@folder:` (no partial) — completion text must keep the
`@folder:` prefix even though the previous implementation auto-rewrote
it to `@file:` for non-dir entries.
"""
monkeypatch.chdir(tmp_path)
texts = [t for t, _ in _run(tmp_path, "@folder:")]
assert texts, "expected at least one directory completion"
for t in texts:
assert t.startswith("@folder:"), f"prefix leaked: {t}"
def test_at_folder_bare_without_colon_lists_directories(tmp_path, monkeypatch):
"""Typing `@folder` alone (no colon yet) should surface directories so
users don't need to first accept the static `@folder:` hint before
seeing what they're picking from.
"""
monkeypatch.chdir(tmp_path)
texts = [t for t, _ in _run(tmp_path, "@folder")]
assert any(t == "@folder:src/" for t in texts), texts
assert any(t == "@folder:docs/" for t in texts), texts
assert not any(t == "@folder:readme.md" for t in texts)
def test_at_file_bare_without_colon_lists_files(tmp_path, monkeypatch):
monkeypatch.chdir(tmp_path)
texts = [t for t, _ in _run(tmp_path, "@file")]
assert any(t == "@file:readme.md" for t in texts), texts
assert not any(t == "@file:src/" for t in texts)

View file

@ -376,7 +376,6 @@ class TestLoginNousSkipKeepsCurrent:
lambda *a, **kw: prompt_returns,
)
monkeypatch.setattr(models_mod, "get_pricing_for_provider", lambda p: {})
monkeypatch.setattr(models_mod, "filter_nous_free_models", lambda ids, p: ids)
monkeypatch.setattr(models_mod, "check_nous_free_tier", lambda: None)
monkeypatch.setattr(
models_mod, "partition_nous_models_by_tier",

View file

@ -0,0 +1,36 @@
"""Regression tests for removed dead config keys.
This file guards against accidental re-introduction of config keys that were
documented or declared at some point but never actually wired up to read code.
Future dead-config regressions can accumulate here.
"""
import inspect
def test_delegation_default_toolsets_removed_from_cli_config():
"""delegation.default_toolsets was dead config — never read by
_load_config() or anywhere else. Removed.
Guards against accidental re-introduction in cli.py's CLI_CONFIG default
dict. If this test fails, someone re-added the key without wiring it up
to _load_config() in tools/delegate_tool.py.
We inspect the source of load_cli_config() instead of asserting on the
runtime CLI_CONFIG dict because CLI_CONFIG is populated by deep-merging
the user's ~/.hermes/config.yaml over the defaults (cli.py:359-366).
A contributor who still has the legacy key set in their own config
would cause a false failure, and HERMES_HOME patching via conftest
doesn't help because cli._hermes_home is frozen at module import time
(cli.py:76) before any autouse fixture can fire. Source inspection
sidesteps all of that: it tests the defaults literal directly.
"""
from cli import load_cli_config
source = inspect.getsource(load_cli_config)
assert '"default_toolsets"' not in source, (
"delegation.default_toolsets was removed because it was never read. "
"Do not re-add it to cli.py's CLI_CONFIG default dict; "
"use tools/delegate_tool.py's DEFAULT_TOOLSETS module constant or "
"wire a new config key through _load_config()."
)

View file

@ -0,0 +1,174 @@
"""Tests for plugin image_gen providers injecting themselves into the picker.
Covers `_plugin_image_gen_providers`, `_visible_providers`, and
`_toolset_needs_configuration_prompt` handling of plugin providers.
"""
from __future__ import annotations
import pytest
from agent import image_gen_registry
from agent.image_gen_provider import ImageGenProvider
class _FakeProvider(ImageGenProvider):
def __init__(self, name: str, available: bool = True, schema=None, models=None):
self._name = name
self._available = available
self._schema = schema or {
"name": name.title(),
"badge": "test",
"tag": f"{name} test tag",
"env_vars": [{"key": f"{name.upper()}_API_KEY", "prompt": f"{name} key"}],
}
self._models = models or [
{"id": f"{name}-model-v1", "display": f"{name} v1",
"speed": "~5s", "strengths": "test", "price": "$"},
]
@property
def name(self) -> str:
return self._name
def is_available(self) -> bool:
return self._available
def list_models(self):
return list(self._models)
def default_model(self):
return self._models[0]["id"] if self._models else None
def get_setup_schema(self):
return dict(self._schema)
def generate(self, prompt, aspect_ratio="landscape", **kw):
return {"success": True, "image": f"{self._name}://{prompt}"}
@pytest.fixture(autouse=True)
def _reset_registry():
image_gen_registry._reset_for_tests()
yield
image_gen_registry._reset_for_tests()
class TestPluginPickerInjection:
def test_plugin_providers_returns_registered(self, monkeypatch):
from hermes_cli import tools_config
image_gen_registry.register_provider(_FakeProvider("myimg"))
rows = tools_config._plugin_image_gen_providers()
names = [r["name"] for r in rows]
plugin_names = [r.get("image_gen_plugin_name") for r in rows]
assert "Myimg" in names
assert "myimg" in plugin_names
def test_fal_skipped_to_avoid_duplicate(self, monkeypatch):
from hermes_cli import tools_config
# Simulate a FAL plugin being registered — the picker already has
# hardcoded FAL rows in TOOL_CATEGORIES, so plugin-FAL must be
# skipped to avoid showing FAL twice.
image_gen_registry.register_provider(_FakeProvider("fal"))
image_gen_registry.register_provider(_FakeProvider("openai"))
rows = tools_config._plugin_image_gen_providers()
names = [r.get("image_gen_plugin_name") for r in rows]
assert "fal" not in names
assert "openai" in names
def test_visible_providers_includes_plugins_for_image_gen(self, monkeypatch):
from hermes_cli import tools_config
image_gen_registry.register_provider(_FakeProvider("someimg"))
cat = tools_config.TOOL_CATEGORIES["image_gen"]
visible = tools_config._visible_providers(cat, {})
plugin_names = [p.get("image_gen_plugin_name") for p in visible if p.get("image_gen_plugin_name")]
assert "someimg" in plugin_names
def test_visible_providers_does_not_inject_into_other_categories(self, monkeypatch):
from hermes_cli import tools_config
image_gen_registry.register_provider(_FakeProvider("someimg"))
# Browser category must NOT see image_gen plugins.
browser = tools_config.TOOL_CATEGORIES["browser"]
visible = tools_config._visible_providers(browser, {})
assert all(p.get("image_gen_plugin_name") is None for p in visible)
class TestPluginCatalog:
def test_plugin_catalog_returns_models(self):
from hermes_cli import tools_config
image_gen_registry.register_provider(_FakeProvider("catimg"))
catalog, default = tools_config._plugin_image_gen_catalog("catimg")
assert "catimg-model-v1" in catalog
assert default == "catimg-model-v1"
def test_plugin_catalog_empty_for_unknown(self):
from hermes_cli import tools_config
catalog, default = tools_config._plugin_image_gen_catalog("does-not-exist")
assert catalog == {}
assert default is None
class TestConfigPrompt:
def test_image_gen_satisfied_by_plugin_provider(self, monkeypatch, tmp_path):
"""When a plugin provider reports is_available(), the picker should
not force a setup prompt on the user."""
from hermes_cli import tools_config
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("FAL_KEY", raising=False)
image_gen_registry.register_provider(_FakeProvider("avail-img", available=True))
assert tools_config._toolset_needs_configuration_prompt("image_gen", {}) is False
def test_image_gen_still_prompts_when_nothing_available(self, monkeypatch, tmp_path):
from hermes_cli import tools_config
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
monkeypatch.delenv("FAL_KEY", raising=False)
image_gen_registry.register_provider(_FakeProvider("unavail-img", available=False))
assert tools_config._toolset_needs_configuration_prompt("image_gen", {}) is True
class TestConfigWriting:
def test_picking_plugin_provider_writes_provider_and_model(self, monkeypatch, tmp_path):
"""When a user picks a plugin-backed image_gen provider with no
env vars needed, ``_configure_provider`` should write both
``image_gen.provider`` and ``image_gen.model``."""
from hermes_cli import tools_config
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
image_gen_registry.register_provider(_FakeProvider("noenv", schema={
"name": "NoEnv",
"badge": "free",
"tag": "",
"env_vars": [],
}))
# Stub out the interactive model picker — no TTY in tests.
monkeypatch.setattr(tools_config, "_prompt_choice", lambda *a, **kw: 0)
config: dict = {}
provider_row = {
"name": "NoEnv",
"env_vars": [],
"image_gen_plugin_name": "noenv",
}
tools_config._configure_provider(provider_row, config)
assert config["image_gen"]["provider"] == "noenv"
assert config["image_gen"]["model"] == "noenv-model-v1"

View file

@ -4,7 +4,6 @@ from unittest.mock import patch, MagicMock
from hermes_cli.models import (
OPENROUTER_MODELS, fetch_openrouter_models, model_ids, detect_provider_for_model,
filter_nous_free_models, _NOUS_ALLOWED_FREE_MODELS,
is_nous_free_tier, partition_nous_models_by_tier,
check_nous_free_tier, _FREE_TIER_CACHE_TTL,
)
@ -293,89 +292,6 @@ class TestDetectProviderForModel:
assert result[0] not in ("nous",) # nous has claude models but shouldn't be suggested
class TestFilterNousFreeModels:
"""Tests for filter_nous_free_models — Nous Portal free-model policy."""
_PAID = {"prompt": "0.000003", "completion": "0.000015"}
_FREE = {"prompt": "0", "completion": "0"}
def test_paid_models_kept(self):
"""Regular paid models pass through unchanged."""
models = ["anthropic/claude-opus-4.6", "openai/gpt-5.4"]
pricing = {m: self._PAID for m in models}
assert filter_nous_free_models(models, pricing) == models
def test_free_non_allowlist_models_removed(self):
"""Free models NOT in the allowlist are filtered out."""
models = ["anthropic/claude-opus-4.6", "arcee-ai/trinity-large-preview:free"]
pricing = {
"anthropic/claude-opus-4.6": self._PAID,
"arcee-ai/trinity-large-preview:free": self._FREE,
}
result = filter_nous_free_models(models, pricing)
assert result == ["anthropic/claude-opus-4.6"]
def test_allowlist_model_kept_when_free(self):
"""Allowlist models are kept when they report as free."""
models = ["anthropic/claude-opus-4.6", "xiaomi/mimo-v2-pro"]
pricing = {
"anthropic/claude-opus-4.6": self._PAID,
"xiaomi/mimo-v2-pro": self._FREE,
}
result = filter_nous_free_models(models, pricing)
assert result == ["anthropic/claude-opus-4.6", "xiaomi/mimo-v2-pro"]
def test_allowlist_model_removed_when_paid(self):
"""Allowlist models are removed when they are NOT free."""
models = ["anthropic/claude-opus-4.6", "xiaomi/mimo-v2-pro"]
pricing = {
"anthropic/claude-opus-4.6": self._PAID,
"xiaomi/mimo-v2-pro": self._PAID,
}
result = filter_nous_free_models(models, pricing)
assert result == ["anthropic/claude-opus-4.6"]
def test_no_pricing_returns_all(self):
"""When pricing data is unavailable, all models pass through."""
models = ["anthropic/claude-opus-4.6", "nvidia/nemotron-3-super-120b-a12b:free"]
assert filter_nous_free_models(models, {}) == models
def test_model_with_no_pricing_entry_treated_as_paid(self):
"""A model missing from the pricing dict is kept (assumed paid)."""
models = ["anthropic/claude-opus-4.6", "openai/gpt-5.4"]
pricing = {"anthropic/claude-opus-4.6": self._PAID} # gpt-5.4 not in pricing
result = filter_nous_free_models(models, pricing)
assert result == models
def test_mixed_scenario(self):
"""End-to-end: mix of paid, free-allowed, free-disallowed, allowlist-not-free."""
models = [
"anthropic/claude-opus-4.6", # paid, not allowlist → keep
"nvidia/nemotron-3-super-120b-a12b:free", # free, not allowlist → drop
"xiaomi/mimo-v2-pro", # free, allowlist → keep
"xiaomi/mimo-v2-omni", # paid, allowlist → drop
"openai/gpt-5.4", # paid, not allowlist → keep
]
pricing = {
"anthropic/claude-opus-4.6": self._PAID,
"nvidia/nemotron-3-super-120b-a12b:free": self._FREE,
"xiaomi/mimo-v2-pro": self._FREE,
"xiaomi/mimo-v2-omni": self._PAID,
"openai/gpt-5.4": self._PAID,
}
result = filter_nous_free_models(models, pricing)
assert result == [
"anthropic/claude-opus-4.6",
"xiaomi/mimo-v2-pro",
"openai/gpt-5.4",
]
def test_allowlist_contains_expected_models(self):
"""Sanity: the allowlist has the models we expect."""
assert "xiaomi/mimo-v2-pro" in _NOUS_ALLOWED_FREE_MODELS
assert "xiaomi/mimo-v2-omni" in _NOUS_ALLOWED_FREE_MODELS
class TestIsNousFreeTier:
"""Tests for is_nous_free_tier — account tier detection."""
@ -501,3 +417,190 @@ class TestCheckNousFreeTierCache:
def test_cache_ttl_is_short(self):
"""TTL should be short enough to catch upgrades quickly (<=5 min)."""
assert _FREE_TIER_CACHE_TTL <= 300
class TestNousRecommendedModels:
"""Tests for fetch_nous_recommended_models + get_nous_recommended_aux_model."""
_SAMPLE_PAYLOAD = {
"paidRecommendedModels": [],
"freeRecommendedModels": [],
"paidRecommendedCompactionModel": None,
"paidRecommendedVisionModel": None,
"freeRecommendedCompactionModel": {
"modelName": "google/gemini-3-flash-preview",
"displayName": "Google: Gemini 3 Flash Preview",
},
"freeRecommendedVisionModel": {
"modelName": "google/gemini-3-flash-preview",
"displayName": "Google: Gemini 3 Flash Preview",
},
}
def setup_method(self):
_models_mod._nous_recommended_cache.clear()
def teardown_method(self):
_models_mod._nous_recommended_cache.clear()
def _mock_urlopen(self, payload):
"""Return a context-manager mock mimicking urllib.request.urlopen()."""
import json as _json
response = MagicMock()
response.read.return_value = _json.dumps(payload).encode()
cm = MagicMock()
cm.__enter__.return_value = response
cm.__exit__.return_value = False
return cm
def test_fetch_caches_per_portal_url(self):
from hermes_cli.models import fetch_nous_recommended_models
mock_cm = self._mock_urlopen(self._SAMPLE_PAYLOAD)
with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen:
a = fetch_nous_recommended_models("https://portal.example.com")
b = fetch_nous_recommended_models("https://portal.example.com")
assert a == self._SAMPLE_PAYLOAD
assert b == self._SAMPLE_PAYLOAD
assert mock_urlopen.call_count == 1 # second call served from cache
def test_fetch_cache_is_keyed_per_portal(self):
from hermes_cli.models import fetch_nous_recommended_models
mock_cm = self._mock_urlopen(self._SAMPLE_PAYLOAD)
with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen:
fetch_nous_recommended_models("https://portal.example.com")
fetch_nous_recommended_models("https://portal.staging-nousresearch.com")
assert mock_urlopen.call_count == 2 # different portals → separate fetches
def test_fetch_returns_empty_on_network_failure(self):
from hermes_cli.models import fetch_nous_recommended_models
with patch("urllib.request.urlopen", side_effect=OSError("boom")):
result = fetch_nous_recommended_models("https://portal.example.com")
assert result == {}
def test_fetch_force_refresh_bypasses_cache(self):
from hermes_cli.models import fetch_nous_recommended_models
mock_cm = self._mock_urlopen(self._SAMPLE_PAYLOAD)
with patch("urllib.request.urlopen", return_value=mock_cm) as mock_urlopen:
fetch_nous_recommended_models("https://portal.example.com")
fetch_nous_recommended_models("https://portal.example.com", force_refresh=True)
assert mock_urlopen.call_count == 2
def test_get_aux_model_returns_vision_recommendation(self):
from hermes_cli.models import get_nous_recommended_aux_model
with patch(
"hermes_cli.models.fetch_nous_recommended_models",
return_value=self._SAMPLE_PAYLOAD,
):
# Free tier → free vision recommendation.
model = get_nous_recommended_aux_model(vision=True, free_tier=True)
assert model == "google/gemini-3-flash-preview"
def test_get_aux_model_returns_compaction_recommendation(self):
from hermes_cli.models import get_nous_recommended_aux_model
payload = dict(self._SAMPLE_PAYLOAD)
payload["freeRecommendedCompactionModel"] = {"modelName": "minimax/minimax-m2.7"}
with patch(
"hermes_cli.models.fetch_nous_recommended_models",
return_value=payload,
):
model = get_nous_recommended_aux_model(vision=False, free_tier=True)
assert model == "minimax/minimax-m2.7"
def test_get_aux_model_returns_none_when_field_null(self):
from hermes_cli.models import get_nous_recommended_aux_model
payload = dict(self._SAMPLE_PAYLOAD)
payload["freeRecommendedCompactionModel"] = None
with patch(
"hermes_cli.models.fetch_nous_recommended_models",
return_value=payload,
):
model = get_nous_recommended_aux_model(vision=False, free_tier=True)
assert model is None
def test_get_aux_model_returns_none_on_empty_payload(self):
from hermes_cli.models import get_nous_recommended_aux_model
with patch("hermes_cli.models.fetch_nous_recommended_models", return_value={}):
assert get_nous_recommended_aux_model(vision=False, free_tier=True) is None
assert get_nous_recommended_aux_model(vision=True, free_tier=False) is None
def test_get_aux_model_returns_none_when_modelname_blank(self):
from hermes_cli.models import get_nous_recommended_aux_model
payload = {"freeRecommendedCompactionModel": {"modelName": " "}}
with patch(
"hermes_cli.models.fetch_nous_recommended_models",
return_value=payload,
):
assert get_nous_recommended_aux_model(vision=False, free_tier=True) is None
def test_paid_tier_prefers_paid_recommendation(self):
"""Paid-tier users should get the paid model when it's populated."""
from hermes_cli.models import get_nous_recommended_aux_model
payload = {
"paidRecommendedCompactionModel": {"modelName": "anthropic/claude-opus-4.7"},
"freeRecommendedCompactionModel": {"modelName": "google/gemini-3-flash-preview"},
"paidRecommendedVisionModel": {"modelName": "openai/gpt-5.4"},
"freeRecommendedVisionModel": {"modelName": "google/gemini-3-flash-preview"},
}
with patch("hermes_cli.models.fetch_nous_recommended_models", return_value=payload):
text = get_nous_recommended_aux_model(vision=False, free_tier=False)
vision = get_nous_recommended_aux_model(vision=True, free_tier=False)
assert text == "anthropic/claude-opus-4.7"
assert vision == "openai/gpt-5.4"
def test_paid_tier_falls_back_to_free_when_paid_is_null(self):
"""If the Portal returns null for the paid field, fall back to free."""
from hermes_cli.models import get_nous_recommended_aux_model
payload = {
"paidRecommendedCompactionModel": None,
"freeRecommendedCompactionModel": {"modelName": "google/gemini-3-flash-preview"},
"paidRecommendedVisionModel": None,
"freeRecommendedVisionModel": {"modelName": "google/gemini-3-flash-preview"},
}
with patch("hermes_cli.models.fetch_nous_recommended_models", return_value=payload):
text = get_nous_recommended_aux_model(vision=False, free_tier=False)
vision = get_nous_recommended_aux_model(vision=True, free_tier=False)
assert text == "google/gemini-3-flash-preview"
assert vision == "google/gemini-3-flash-preview"
def test_free_tier_never_uses_paid_recommendation(self):
"""Free-tier users must not get paid-only recommendations."""
from hermes_cli.models import get_nous_recommended_aux_model
payload = {
"paidRecommendedCompactionModel": {"modelName": "anthropic/claude-opus-4.7"},
"freeRecommendedCompactionModel": None, # no free recommendation
}
with patch("hermes_cli.models.fetch_nous_recommended_models", return_value=payload):
model = get_nous_recommended_aux_model(vision=False, free_tier=True)
# Free tier must return None — never leak the paid model.
assert model is None
def test_auto_detects_tier_when_not_supplied(self):
"""Default behaviour: call check_nous_free_tier() to pick the tier."""
from hermes_cli.models import get_nous_recommended_aux_model
payload = {
"paidRecommendedCompactionModel": {"modelName": "paid-model"},
"freeRecommendedCompactionModel": {"modelName": "free-model"},
}
with (
patch("hermes_cli.models.fetch_nous_recommended_models", return_value=payload),
patch("hermes_cli.models.check_nous_free_tier", return_value=True),
):
assert get_nous_recommended_aux_model(vision=False) == "free-model"
with (
patch("hermes_cli.models.fetch_nous_recommended_models", return_value=payload),
patch("hermes_cli.models.check_nous_free_tier", return_value=False),
):
assert get_nous_recommended_aux_model(vision=False) == "paid-model"
def test_tier_detection_error_defaults_to_paid(self):
"""If tier detection raises, assume paid so we don't downgrade silently."""
from hermes_cli.models import get_nous_recommended_aux_model
payload = {
"paidRecommendedCompactionModel": {"modelName": "paid-model"},
"freeRecommendedCompactionModel": {"modelName": "free-model"},
}
with (
patch("hermes_cli.models.fetch_nous_recommended_models", return_value=payload),
patch("hermes_cli.models.check_nous_free_tier", side_effect=RuntimeError("boom")),
):
assert get_nous_recommended_aux_model(vision=False) == "paid-model"

View file

@ -0,0 +1,357 @@
"""Tests for PR1 pluggable image gen: scanner recursion, kinds, path keys.
Covers ``_scan_directory`` recursion into category namespaces
(``plugins/image_gen/openai/``), ``kind`` parsing, path-derived registry
keys, and the new gate logic (bundled backends auto-load; user backends
still opt-in; exclusive kind skipped; unknown kinds standalone warning).
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict
import pytest
import yaml
from hermes_cli.plugins import PluginManager, PluginManifest
# ── Helpers ────────────────────────────────────────────────────────────────
def _write_plugin(
root: Path,
segments: list[str],
*,
manifest_extra: Dict[str, Any] | None = None,
register_body: str = "pass",
) -> Path:
"""Create a plugin dir at ``root/<segments...>/`` with plugin.yaml + __init__.py.
``segments`` lets tests build both flat (``["my-plugin"]``) and
category-namespaced (``["image_gen", "openai"]``) layouts.
"""
plugin_dir = root
for seg in segments:
plugin_dir = plugin_dir / seg
plugin_dir.mkdir(parents=True, exist_ok=True)
manifest = {
"name": segments[-1],
"version": "0.1.0",
"description": f"Test plugin {'/'.join(segments)}",
}
if manifest_extra:
manifest.update(manifest_extra)
(plugin_dir / "plugin.yaml").write_text(yaml.dump(manifest))
(plugin_dir / "__init__.py").write_text(
f"def register(ctx):\n {register_body}\n"
)
return plugin_dir
def _enable(hermes_home: Path, name: str) -> None:
"""Append ``name`` to ``plugins.enabled`` in ``<hermes_home>/config.yaml``."""
cfg_path = hermes_home / "config.yaml"
cfg: dict = {}
if cfg_path.exists():
try:
cfg = yaml.safe_load(cfg_path.read_text()) or {}
except Exception:
cfg = {}
plugins_cfg = cfg.setdefault("plugins", {})
enabled = plugins_cfg.setdefault("enabled", [])
if isinstance(enabled, list) and name not in enabled:
enabled.append(name)
cfg_path.write_text(yaml.safe_dump(cfg))
# ── Scanner recursion ──────────────────────────────────────────────────────
class TestCategoryNamespaceRecursion:
def test_category_namespace_discovered(self, tmp_path, monkeypatch):
"""``<root>/image_gen/openai/plugin.yaml`` is discovered with key
``image_gen/openai`` when the ``image_gen`` parent has no manifest."""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
user_plugins = hermes_home / "plugins"
_write_plugin(user_plugins, ["image_gen", "openai"])
_enable(hermes_home, "image_gen/openai")
mgr = PluginManager()
mgr.discover_and_load()
assert "image_gen/openai" in mgr._plugins
loaded = mgr._plugins["image_gen/openai"]
assert loaded.manifest.key == "image_gen/openai"
assert loaded.manifest.name == "openai"
assert loaded.enabled is True
def test_flat_plugin_key_matches_name(self, tmp_path, monkeypatch):
"""Flat plugins keep their bare name as the key (back-compat)."""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
user_plugins = hermes_home / "plugins"
_write_plugin(user_plugins, ["my-plugin"])
_enable(hermes_home, "my-plugin")
mgr = PluginManager()
mgr.discover_and_load()
assert "my-plugin" in mgr._plugins
assert mgr._plugins["my-plugin"].manifest.key == "my-plugin"
def test_depth_cap_two(self, tmp_path, monkeypatch):
"""Plugins nested three levels deep are not discovered.
``<root>/a/b/c/plugin.yaml`` should NOT be picked up cap is
two segments.
"""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
user_plugins = hermes_home / "plugins"
_write_plugin(user_plugins, ["a", "b", "c"])
mgr = PluginManager()
mgr.discover_and_load()
non_bundled = [
k for k, p in mgr._plugins.items()
if p.manifest.source != "bundled"
]
assert non_bundled == []
def test_category_dir_with_manifest_is_leaf(self, tmp_path, monkeypatch):
"""If ``image_gen/plugin.yaml`` exists, ``image_gen`` itself IS the
plugin and its children are ignored."""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
user_plugins = hermes_home / "plugins"
# parent has a manifest → stop recursing
_write_plugin(user_plugins, ["image_gen"])
# child also has a manifest — should NOT be found because we stop
# at the parent.
_write_plugin(user_plugins, ["image_gen", "openai"])
_enable(hermes_home, "image_gen")
_enable(hermes_home, "image_gen/openai")
mgr = PluginManager()
mgr.discover_and_load()
# The bundled plugins/image_gen/openai/ exists in the repo — filter
# it out so we're only asserting on the user-dir layout.
user_plugins_in_registry = {
k for k, p in mgr._plugins.items() if p.manifest.source != "bundled"
}
assert "image_gen" in user_plugins_in_registry
assert "image_gen/openai" not in user_plugins_in_registry
# ── Kind parsing ───────────────────────────────────────────────────────────
class TestKindField:
def test_default_kind_is_standalone(self, tmp_path, monkeypatch):
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
_write_plugin(hermes_home / "plugins", ["p1"])
_enable(hermes_home, "p1")
mgr = PluginManager()
mgr.discover_and_load()
assert mgr._plugins["p1"].manifest.kind == "standalone"
@pytest.mark.parametrize("kind", ["backend", "exclusive", "standalone"])
def test_valid_kinds_parsed(self, kind, tmp_path, monkeypatch):
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
_write_plugin(
hermes_home / "plugins",
["p1"],
manifest_extra={"kind": kind},
)
# Not all kinds auto-load, but manifest should parse.
_enable(hermes_home, "p1")
mgr = PluginManager()
mgr.discover_and_load()
assert "p1" in mgr._plugins
assert mgr._plugins["p1"].manifest.kind == kind
def test_unknown_kind_falls_back_to_standalone(self, tmp_path, monkeypatch, caplog):
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
_write_plugin(
hermes_home / "plugins",
["p1"],
manifest_extra={"kind": "bogus"},
)
_enable(hermes_home, "p1")
with caplog.at_level("WARNING"):
mgr = PluginManager()
mgr.discover_and_load()
assert mgr._plugins["p1"].manifest.kind == "standalone"
assert any(
"unknown kind" in rec.getMessage() for rec in caplog.records
)
# ── Gate logic ─────────────────────────────────────────────────────────────
class TestBackendGate:
def test_user_backend_still_gated_by_enabled(self, tmp_path, monkeypatch):
"""User-installed ``kind: backend`` plugins still require opt-in —
they're not trusted by default."""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
user_plugins = hermes_home / "plugins"
_write_plugin(
user_plugins,
["image_gen", "fancy"],
manifest_extra={"kind": "backend"},
)
# Do NOT opt in.
mgr = PluginManager()
mgr.discover_and_load()
loaded = mgr._plugins["image_gen/fancy"]
assert loaded.enabled is False
assert "not enabled" in (loaded.error or "")
def test_user_backend_loads_when_enabled(self, tmp_path, monkeypatch):
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
user_plugins = hermes_home / "plugins"
_write_plugin(
user_plugins,
["image_gen", "fancy"],
manifest_extra={"kind": "backend"},
)
_enable(hermes_home, "image_gen/fancy")
mgr = PluginManager()
mgr.discover_and_load()
assert mgr._plugins["image_gen/fancy"].enabled is True
def test_exclusive_kind_skipped(self, tmp_path, monkeypatch):
"""``kind: exclusive`` plugins are recorded but not loaded — the
category's own discovery system handles them (memory today)."""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
_write_plugin(
hermes_home / "plugins",
["some-backend"],
manifest_extra={"kind": "exclusive"},
)
_enable(hermes_home, "some-backend")
mgr = PluginManager()
mgr.discover_and_load()
loaded = mgr._plugins["some-backend"]
assert loaded.enabled is False
assert "exclusive" in (loaded.error or "")
# ── Bundled backend auto-load (integration with real bundled plugin) ────────
class TestBundledBackendAutoLoad:
def test_bundled_image_gen_openai_autoloads(self, tmp_path, monkeypatch):
"""The bundled ``plugins/image_gen/openai/`` plugin loads without
any opt-in it's ``kind: backend`` and shipped in-repo."""
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
mgr = PluginManager()
mgr.discover_and_load()
assert "image_gen/openai" in mgr._plugins
loaded = mgr._plugins["image_gen/openai"]
assert loaded.manifest.source == "bundled"
assert loaded.manifest.kind == "backend"
assert loaded.enabled is True, f"error: {loaded.error}"
# ── PluginContext.register_image_gen_provider ───────────────────────────────
class TestRegisterImageGenProvider:
def test_accepts_valid_provider(self, tmp_path, monkeypatch):
from agent import image_gen_registry
from agent.image_gen_provider import ImageGenProvider
image_gen_registry._reset_for_tests()
class FakeProvider(ImageGenProvider):
@property
def name(self) -> str:
return "fake-test"
def generate(self, prompt, aspect_ratio="landscape", **kw):
return {"success": True, "image": "test://fake"}
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
plugin_dir = _write_plugin(
hermes_home / "plugins",
["my-img-plugin"],
register_body=(
"from agent.image_gen_provider import ImageGenProvider\n"
" class P(ImageGenProvider):\n"
" @property\n"
" def name(self): return 'fake-ctx'\n"
" def generate(self, prompt, aspect_ratio='landscape', **kw):\n"
" return {'success': True, 'image': 'x://y'}\n"
" ctx.register_image_gen_provider(P())"
),
)
_enable(hermes_home, "my-img-plugin")
mgr = PluginManager()
mgr.discover_and_load()
assert mgr._plugins["my-img-plugin"].enabled is True
assert image_gen_registry.get_provider("fake-ctx") is not None
image_gen_registry._reset_for_tests()
def test_rejects_non_provider(self, tmp_path, monkeypatch, caplog):
from agent import image_gen_registry
image_gen_registry._reset_for_tests()
import os
hermes_home = Path(os.environ["HERMES_HOME"]) # set by hermetic conftest fixture
_write_plugin(
hermes_home / "plugins",
["bad-img-plugin"],
register_body="ctx.register_image_gen_provider('not a provider')",
)
_enable(hermes_home, "bad-img-plugin")
with caplog.at_level("WARNING"):
mgr = PluginManager()
mgr.discover_and_load()
# Plugin loaded (register returned normally) but nothing was
# registered in the provider registry.
assert mgr._plugins["bad-img-plugin"].enabled is True
assert image_gen_registry.get_provider("not a provider") is None
image_gen_registry._reset_for_tests()