feat(plugins): pluggable image_gen backends + OpenAI provider (#13799)

* feat(plugins): pluggable image_gen backends + OpenAI provider

Adds a ImageGenProvider ABC so image generation backends register as
bundled plugins under `plugins/image_gen/<name>/`. The plugin scanner
gains three primitives to make this work generically:

- `kind:` manifest field (`standalone` | `backend` | `exclusive`).
  Bundled `kind: backend` plugins auto-load — no `plugins.enabled`
  incantation. User-installed backends stay opt-in.
- Path-derived keys: `plugins/image_gen/openai/` gets key
  `image_gen/openai`, so a future `tts/openai` cannot collide.
- Depth-2 recursion into category namespaces (parent dirs without a
  `plugin.yaml` of their own).

Includes `OpenAIImageGenProvider` as the first consumer (gpt-image-1.5
default, plus gpt-image-1, gpt-image-1-mini, DALL-E 3/2). Base64
responses save to `$HERMES_HOME/cache/images/`; URL responses pass
through.

FAL stays in-tree for this PR — a follow-up ports it into
`plugins/image_gen/fal/` so the in-tree `image_generation_tool.py`
slims down. The dispatch shim in `_handle_image_generate` only fires
when `image_gen.provider` is explicitly set to a non-FAL value, so
existing FAL setups are untouched.

- 41 unit tests (scanner recursion, kind parsing, gate logic,
  registry, OpenAI payload shapes)
- E2E smoke verified: bundled plugin autoloads, registers, and
  `_handle_image_generate` routes to OpenAI when configured

* fix(image_gen/openai): don't send response_format to gpt-image-*

The live API rejects it: 'Unknown parameter: response_format'
(verified 2026-04-21 with gpt-image-1.5). gpt-image-* models return
b64_json unconditionally, so the parameter was both unnecessary and
actively broken.

* feat(image_gen/openai): gpt-image-2 only, drop legacy catalog

gpt-image-2 is the latest/best OpenAI image model (released 2026-04-21)
and there's no reason to expose the older gpt-image-1.5 / gpt-image-1 /
dall-e-3 / dall-e-2 alongside it — slower, lower quality, or awkward
(dall-e-2 squares only). Trim the catalog down to a single model.

Live-verified end-to-end: landscape 1536x1024 render of a Moog-style
synth matches prompt exactly, 2.4MB PNG saved to cache.

* feat(image_gen/openai): expose gpt-image-2 as three quality tiers

Users pick speed/fidelity via the normal model picker instead of a
hidden quality knob. All three tier IDs resolve to the single underlying
gpt-image-2 API model with a different quality parameter:

  gpt-image-2-low     ~15s   fast iteration
  gpt-image-2-medium  ~40s   default
  gpt-image-2-high    ~2min  highest fidelity

Live-measured on OpenAI's API today: 15.4s / 40.8s / 116.9s for the
same 1024x1024 prompt.

Config:
  image_gen.openai.model: gpt-image-2-high
  # or
  image_gen.model: gpt-image-2-low
  # or env var for scripts/tests
  OPENAI_IMAGE_MODEL=gpt-image-2-medium

Live-verified end-to-end with the low tier: 18.8s landscape render of a
golden retriever in wildflowers, vision-confirmed exact match.

* feat(tools_config): plugin image_gen providers inject themselves into picker

'hermes tools' → Image Generation now shows plugin-registered backends
alongside Nous Subscription and FAL.ai without tools_config.py needing
to know about them. OpenAI appears as a third option today; future
backends appear automatically as they're added.

Mechanism:
- ImageGenProvider gains an optional get_setup_schema() hook
  (name, badge, tag, env_vars). Default derived from display_name.
- tools_config._plugin_image_gen_providers() pulls the schemas from
  every registered non-FAL plugin provider.
- _visible_providers() appends those rows when rendering the Image
  Generation category.
- _configure_provider() handles the new image_gen_plugin_name marker:
  writes image_gen.provider and routes to the plugin's list_models()
  catalog for the model picker.
- _toolset_needs_configuration_prompt('image_gen') stops demanding a
  FAL key when any plugin provider reports is_available().

FAL is skipped in the plugin path because it already has hardcoded
TOOL_CATEGORIES rows — when it gets ported to a plugin in a follow-up
PR the hardcoded rows go away and it surfaces through the same path
as OpenAI.

Verified live: picker shows Nous Subscription / FAL.ai / OpenAI.
Picking OpenAI prompts for OPENAI_API_KEY, then shows the
gpt-image-2-low/medium/high model picker sourced from the plugin.

397 tests pass across plugins/, tools_config, registry, and picker.

* fix(image_gen): close final gaps for plugin-backend parity with FAL

Two small places that still hardcoded FAL:

- hermes_cli/setup.py status line: an OpenAI-only setup showed
  'Image Generation: missing FAL_KEY'. Now probes plugin providers
  and reports '(OpenAI)' when one is_available() — or falls back to
  'missing FAL_KEY or OPENAI_API_KEY' if nothing is configured.

- image_generate tool schema description: said 'using FAL.ai, default
  FLUX 2 Klein 9B'. Rewrote provider-neutral — 'backend and model are
  user-configured' — and notes the 'image' field can be a URL or an
  absolute path, which the gateway delivers either way via
  extract_local_files().
This commit is contained in:
Teknium 2026-04-21 21:30:10 -07:00 committed by GitHub
parent d1acf17773
commit ff9752410a
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 2122 additions and 67 deletions

View file

@ -0,0 +1,111 @@
"""Tests for agent/image_gen_registry.py — provider registration & active lookup."""
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):
self._name = name
self._available = available
@property
def name(self) -> str:
return self._name
def is_available(self) -> bool:
return self._available
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 TestRegisterProvider:
def test_register_and_lookup(self):
provider = _FakeProvider("fake")
image_gen_registry.register_provider(provider)
assert image_gen_registry.get_provider("fake") is provider
def test_rejects_non_provider(self):
with pytest.raises(TypeError):
image_gen_registry.register_provider("not a provider") # type: ignore[arg-type]
def test_rejects_empty_name(self):
class Empty(ImageGenProvider):
@property
def name(self) -> str:
return ""
def generate(self, prompt, aspect_ratio="landscape", **kw):
return {}
with pytest.raises(ValueError):
image_gen_registry.register_provider(Empty())
def test_reregister_overwrites(self):
a = _FakeProvider("same")
b = _FakeProvider("same")
image_gen_registry.register_provider(a)
image_gen_registry.register_provider(b)
assert image_gen_registry.get_provider("same") is b
def test_list_is_sorted(self):
image_gen_registry.register_provider(_FakeProvider("zeta"))
image_gen_registry.register_provider(_FakeProvider("alpha"))
names = [p.name for p in image_gen_registry.list_providers()]
assert names == ["alpha", "zeta"]
class TestGetActiveProvider:
def test_single_provider_autoresolves(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
image_gen_registry.register_provider(_FakeProvider("solo"))
active = image_gen_registry.get_active_provider()
assert active is not None and active.name == "solo"
def test_fal_preferred_on_multi_without_config(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
image_gen_registry.register_provider(_FakeProvider("fal"))
image_gen_registry.register_provider(_FakeProvider("openai"))
active = image_gen_registry.get_active_provider()
assert active is not None and active.name == "fal"
def test_explicit_config_wins(self, tmp_path, monkeypatch):
import yaml
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text(
yaml.safe_dump({"image_gen": {"provider": "openai"}})
)
image_gen_registry.register_provider(_FakeProvider("fal"))
image_gen_registry.register_provider(_FakeProvider("openai"))
active = image_gen_registry.get_active_provider()
assert active is not None and active.name == "openai"
def test_missing_configured_provider_falls_back(self, tmp_path, monkeypatch):
import yaml
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
(tmp_path / "config.yaml").write_text(
yaml.safe_dump({"image_gen": {"provider": "replicate"}})
)
# Only FAL is registered — configured provider doesn't exist
image_gen_registry.register_provider(_FakeProvider("fal"))
active = image_gen_registry.get_active_provider()
# Falls back to FAL preference (legacy default) rather than None
assert active is not None and active.name == "fal"
def test_none_when_empty(self, tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
assert image_gen_registry.get_active_provider() is None

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

@ -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()

View file

View file

@ -0,0 +1,243 @@
"""Tests for the bundled OpenAI image_gen plugin (gpt-image-2, three tiers)."""
from __future__ import annotations
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
import plugins.image_gen.openai as openai_plugin
# 1×1 transparent PNG — valid bytes for save_b64_image()
_PNG_HEX = (
"89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4"
"890000000d49444154789c6300010000000500010d0a2db40000000049454e44"
"ae426082"
)
def _b64_png() -> str:
import base64
return base64.b64encode(bytes.fromhex(_PNG_HEX)).decode()
def _fake_response(*, b64=None, url=None, revised_prompt=None):
item = SimpleNamespace(b64_json=b64, url=url, revised_prompt=revised_prompt)
return SimpleNamespace(data=[item])
@pytest.fixture(autouse=True)
def _tmp_hermes_home(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
yield tmp_path
@pytest.fixture
def provider(monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "test-key")
return openai_plugin.OpenAIImageGenProvider()
def _patched_openai(fake_client: MagicMock):
fake_openai = MagicMock()
fake_openai.OpenAI.return_value = fake_client
return patch.dict("sys.modules", {"openai": fake_openai})
# ── Metadata ────────────────────────────────────────────────────────────────
class TestMetadata:
def test_name(self, provider):
assert provider.name == "openai"
def test_default_model(self, provider):
assert provider.default_model() == "gpt-image-2-medium"
def test_list_models_three_tiers(self, provider):
ids = [m["id"] for m in provider.list_models()]
assert ids == ["gpt-image-2-low", "gpt-image-2-medium", "gpt-image-2-high"]
def test_catalog_entries_have_display_speed_strengths(self, provider):
for entry in provider.list_models():
assert entry["display"].startswith("GPT Image 2")
assert entry["speed"]
assert entry["strengths"]
# ── Availability ────────────────────────────────────────────────────────────
class TestAvailability:
def test_no_api_key_unavailable(self, monkeypatch):
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
assert openai_plugin.OpenAIImageGenProvider().is_available() is False
def test_api_key_set_available(self, monkeypatch):
monkeypatch.setenv("OPENAI_API_KEY", "test")
assert openai_plugin.OpenAIImageGenProvider().is_available() is True
# ── Model resolution ────────────────────────────────────────────────────────
class TestModelResolution:
def test_default_is_medium(self):
model_id, meta = openai_plugin._resolve_model()
assert model_id == "gpt-image-2-medium"
assert meta["quality"] == "medium"
def test_env_var_override(self, monkeypatch):
monkeypatch.setenv("OPENAI_IMAGE_MODEL", "gpt-image-2-high")
model_id, meta = openai_plugin._resolve_model()
assert model_id == "gpt-image-2-high"
assert meta["quality"] == "high"
def test_env_var_unknown_falls_back(self, monkeypatch):
monkeypatch.setenv("OPENAI_IMAGE_MODEL", "bogus-tier")
model_id, _ = openai_plugin._resolve_model()
assert model_id == openai_plugin.DEFAULT_MODEL
def test_config_openai_model(self, tmp_path):
import yaml
(tmp_path / "config.yaml").write_text(
yaml.safe_dump({"image_gen": {"openai": {"model": "gpt-image-2-low"}}})
)
model_id, meta = openai_plugin._resolve_model()
assert model_id == "gpt-image-2-low"
assert meta["quality"] == "low"
def test_config_top_level_model(self, tmp_path):
"""``image_gen.model: gpt-image-2-high`` also works (top-level)."""
import yaml
(tmp_path / "config.yaml").write_text(
yaml.safe_dump({"image_gen": {"model": "gpt-image-2-high"}})
)
model_id, meta = openai_plugin._resolve_model()
assert model_id == "gpt-image-2-high"
assert meta["quality"] == "high"
# ── Generate ────────────────────────────────────────────────────────────────
class TestGenerate:
def test_empty_prompt_rejected(self, provider):
result = provider.generate("", aspect_ratio="square")
assert result["success"] is False
assert result["error_type"] == "invalid_argument"
def test_missing_api_key(self, monkeypatch):
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
result = openai_plugin.OpenAIImageGenProvider().generate("a cat")
assert result["success"] is False
assert result["error_type"] == "auth_required"
def test_b64_saves_to_cache(self, provider, tmp_path):
import base64
png_bytes = bytes.fromhex(_PNG_HEX)
fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(b64=_b64_png())
with _patched_openai(fake_client):
result = provider.generate("a cat", aspect_ratio="landscape")
assert result["success"] is True
assert result["model"] == "gpt-image-2-medium"
assert result["aspect_ratio"] == "landscape"
assert result["provider"] == "openai"
assert result["quality"] == "medium"
saved = Path(result["image"])
assert saved.exists()
assert saved.parent == tmp_path / "cache" / "images"
assert saved.read_bytes() == png_bytes
call_kwargs = fake_client.images.generate.call_args.kwargs
# All tiers hit the single underlying API model.
assert call_kwargs["model"] == "gpt-image-2"
assert call_kwargs["quality"] == "medium"
assert call_kwargs["size"] == "1536x1024"
# gpt-image-2 rejects response_format — we must NOT send it.
assert "response_format" not in call_kwargs
@pytest.mark.parametrize("tier,expected_quality", [
("gpt-image-2-low", "low"),
("gpt-image-2-medium", "medium"),
("gpt-image-2-high", "high"),
])
def test_tier_maps_to_quality(self, provider, monkeypatch, tier, expected_quality):
monkeypatch.setenv("OPENAI_IMAGE_MODEL", tier)
fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(b64=_b64_png())
with _patched_openai(fake_client):
result = provider.generate("a cat")
assert result["model"] == tier
assert result["quality"] == expected_quality
assert fake_client.images.generate.call_args.kwargs["quality"] == expected_quality
# Always the same underlying API model regardless of tier.
assert fake_client.images.generate.call_args.kwargs["model"] == "gpt-image-2"
@pytest.mark.parametrize("aspect,expected_size", [
("landscape", "1536x1024"),
("square", "1024x1024"),
("portrait", "1024x1536"),
])
def test_aspect_ratio_mapping(self, provider, aspect, expected_size):
fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(b64=_b64_png())
with _patched_openai(fake_client):
provider.generate("a cat", aspect_ratio=aspect)
assert fake_client.images.generate.call_args.kwargs["size"] == expected_size
def test_revised_prompt_passed_through(self, provider):
fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(
b64=_b64_png(), revised_prompt="A photo of a cat",
)
with _patched_openai(fake_client):
result = provider.generate("a cat")
assert result["revised_prompt"] == "A photo of a cat"
def test_api_error_returns_error_response(self, provider):
fake_client = MagicMock()
fake_client.images.generate.side_effect = RuntimeError("boom")
with _patched_openai(fake_client):
result = provider.generate("a cat")
assert result["success"] is False
assert result["error_type"] == "api_error"
assert "boom" in result["error"]
def test_empty_response_data(self, provider):
fake_client = MagicMock()
fake_client.images.generate.return_value = SimpleNamespace(data=[])
with _patched_openai(fake_client):
result = provider.generate("a cat")
assert result["success"] is False
assert result["error_type"] == "empty_response"
def test_url_fallback_if_api_changes(self, provider):
"""Defensive: if OpenAI ever returns URL instead of b64, pass through."""
fake_client = MagicMock()
fake_client.images.generate.return_value = _fake_response(
b64=None, url="https://example.com/img.png",
)
with _patched_openai(fake_client):
result = provider.generate("a cat")
assert result["success"] is True
assert result["image"] == "https://example.com/img.png"