mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-26 17:38:36 +00:00
064b6e40c5
The Codex image backend rejected our own request shape for every account, and we then translated that rejection into "Image generation is not enabled for the current Codex account. Switch the image provider to OpenAI API key, FAL, or xAI." — telling every affected user to abandon a provider that had never actually been tried. That message is why this reads as a setup failure rather than a bug: the wire error was replaced with a confident, wrong diagnosis. Removes the classifier and its exception, so any HTTP failure surfaces verbatim. The paired request-shape fix (previous commit) is what makes the 400 stop happening; this commit makes the next one diagnosable. Also fixes error-body truncation: bodies were head-truncated at 500 chars, and Codex error payloads can carry hundreds of bytes of leading metadata, so the user got a wall of padding and no message. _summarize_error_body() prefers the parsed error.message and falls back to a truncated raw body. Docs: drop the unqualified image-to-image claim for the Codex backend and note that the hosted tool call cannot be forced, so it is best-effort. Verified E2E against a local fake Codex backend: success path writes a real PNG with no tool_choice on the wire; the 400 path now returns api_error carrying "Tool choice 'image_generation' not found in 'tools' parameter" (148 chars) instead of the entitlement message. Sabotage run confirms all 4 regression tests fail when the old behavior is restored. Refs #19505, #49008, #31335.
429 lines
17 KiB
Python
429 lines
17 KiB
Python
"""Tests for the bundled ``openai-codex`` image_gen plugin.
|
||
|
||
Mirrors ``test_openai_provider.py`` but targets the standalone
|
||
Codex/ChatGPT-OAuth-backed provider that uses the Responses
|
||
``image_generation`` tool path instead of the ``images.generate`` REST
|
||
endpoint.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import importlib
|
||
import json
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
# The plugin directory uses a hyphen, which is not a valid Python identifier
|
||
# for the dotted-import form. Load it via importlib so tests don't need to
|
||
# touch sys.path or rename the directory.
|
||
codex_plugin = importlib.import_module("plugins.image_gen.openai-codex")
|
||
|
||
|
||
# 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()
|
||
|
||
|
||
@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):
|
||
# Codex plugin is API-key-independent; clear it to make the test honest.
|
||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||
return codex_plugin.OpenAICodexImageGenProvider()
|
||
|
||
|
||
# ── Metadata ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestMetadata:
|
||
def test_name(self, provider):
|
||
assert provider.name == "openai-codex"
|
||
|
||
def test_display_name(self, provider):
|
||
assert provider.display_name == "OpenAI (Codex auth)"
|
||
|
||
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_setup_schema_has_no_required_env_vars(self, provider):
|
||
schema = provider.get_setup_schema()
|
||
assert schema["env_vars"] == []
|
||
assert schema["badge"] == "free"
|
||
|
||
|
||
# ── Availability ────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestAvailability:
|
||
def test_unavailable_without_codex_token(self, monkeypatch):
|
||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: None)
|
||
assert codex_plugin.OpenAICodexImageGenProvider().is_available() is False
|
||
|
||
def test_available_with_codex_token(self, monkeypatch):
|
||
monkeypatch.delenv("OPENAI_API_KEY", raising=False)
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
assert codex_plugin.OpenAICodexImageGenProvider().is_available() is True
|
||
|
||
def test_openai_api_key_alone_is_not_enough(self, monkeypatch):
|
||
# Codex plugin is intentionally orthogonal to the API-key plugin —
|
||
# the API key alone must NOT make it appear available.
|
||
monkeypatch.setenv("OPENAI_API_KEY", "sk-test")
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: None)
|
||
assert codex_plugin.OpenAICodexImageGenProvider().is_available() is False
|
||
|
||
|
||
# ── Generate ────────────────────────────────────────────────────────────────
|
||
|
||
|
||
class TestGenerate:
|
||
def test_returns_auth_error_without_codex_token(self, provider, monkeypatch):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: None)
|
||
result = provider.generate("a cat")
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "auth_required"
|
||
|
||
def test_returns_invalid_argument_for_empty_prompt(self, provider, monkeypatch):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
result = provider.generate(" ")
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "invalid_argument"
|
||
|
||
def test_generate_uses_codex_stream_path(self, provider, monkeypatch, tmp_path):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
monkeypatch.setattr(codex_plugin, "_collect_image_b64", lambda *a, **kw: _b64_png())
|
||
|
||
result = provider.generate("a cat", aspect_ratio="landscape")
|
||
|
||
assert result["success"] is True
|
||
assert result["model"] == "gpt-image-2-medium"
|
||
assert result["provider"] == "openai-codex"
|
||
assert result["quality"] == "medium"
|
||
|
||
saved = Path(result["image"])
|
||
assert saved.exists()
|
||
assert saved.parent == tmp_path / "cache" / "images"
|
||
# Filename prefix differs from the API-key plugin so cache audits can
|
||
# tell the two backends apart.
|
||
assert saved.name.startswith("openai_codex_")
|
||
|
||
def test_codex_stream_request_shape(self, provider, monkeypatch):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
|
||
captured = {}
|
||
|
||
def _collect(token, *, prompt, size, quality, input_images=None):
|
||
captured.update(codex_plugin._build_responses_payload(
|
||
prompt=prompt,
|
||
size=size,
|
||
quality=quality,
|
||
input_images=input_images,
|
||
))
|
||
return _b64_png()
|
||
|
||
monkeypatch.setattr(codex_plugin, "_collect_image_b64", _collect)
|
||
|
||
result = provider.generate("a cat", aspect_ratio="portrait")
|
||
assert result["success"] is True
|
||
|
||
assert captured["model"] == "gpt-5.5"
|
||
assert captured["store"] is False
|
||
assert captured["input"][0]["type"] == "message"
|
||
assert captured["input"][0]["role"] == "user"
|
||
assert captured["input"][0]["content"][0]["type"] == "input_text"
|
||
# Regression for #19505: the Codex backend 400s on every tool_choice
|
||
# shape we have for the hosted ``image_generation`` tool, so the
|
||
# provider must omit tool_choice entirely and rely on instructions.
|
||
assert "tool_choice" not in captured
|
||
|
||
tool = captured["tools"][0]
|
||
assert tool["type"] == "image_generation"
|
||
assert tool["model"] == "gpt-image-2"
|
||
assert tool["quality"] == "medium"
|
||
assert tool["size"] == "1024x1536"
|
||
assert tool["output_format"] == "png"
|
||
assert tool["background"] == "opaque"
|
||
assert tool["partial_images"] == 1
|
||
|
||
def test_capabilities_advertise_image_inputs(self, provider):
|
||
caps = provider.capabilities()
|
||
assert caps["modalities"] == ["text", "image"]
|
||
assert caps["max_reference_images"] == 16
|
||
|
||
def test_codex_stream_request_includes_source_images(self, provider, monkeypatch, tmp_path):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
image_path = tmp_path / "source.png"
|
||
image_path.write_bytes(bytes.fromhex(_PNG_HEX))
|
||
|
||
captured = {}
|
||
|
||
def _collect(token, *, prompt, size, quality, input_images=None):
|
||
captured.update(codex_plugin._build_responses_payload(
|
||
prompt=prompt,
|
||
size=size,
|
||
quality=quality,
|
||
input_images=input_images,
|
||
))
|
||
return _b64_png()
|
||
|
||
monkeypatch.setattr(codex_plugin, "_collect_image_b64", _collect)
|
||
|
||
result = provider.generate(
|
||
"put this same person in a navy JK uniform",
|
||
aspect_ratio="portrait",
|
||
image_url=str(image_path),
|
||
reference_image_urls=["https://example.com/ref.png"],
|
||
)
|
||
|
||
assert result["success"] is True
|
||
assert result["modality"] == "image"
|
||
assert result["input_image_count"] == 2
|
||
|
||
content = captured["input"][0]["content"]
|
||
assert content[0] == {
|
||
"type": "input_text",
|
||
"text": "put this same person in a navy JK uniform",
|
||
}
|
||
assert content[1]["type"] == "input_image"
|
||
assert content[1]["image_url"].startswith("data:image/png;base64,")
|
||
assert content[2] == {"type": "input_image", "image_url": "https://example.com/ref.png"}
|
||
|
||
def test_generate_clamps_reference_images_to_cap(self, provider, monkeypatch):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
captured = {}
|
||
|
||
def _collect(token, *, prompt, size, quality, input_images=None):
|
||
captured["input_images"] = input_images
|
||
return _b64_png()
|
||
|
||
monkeypatch.setattr(codex_plugin, "_collect_image_b64", _collect)
|
||
|
||
refs = [f"https://example.com/ref-{idx}.png" for idx in range(20)]
|
||
result = provider.generate("combine the references", reference_image_urls=refs)
|
||
|
||
assert result["success"] is True
|
||
assert result["modality"] == "image"
|
||
assert result["input_image_count"] == 16
|
||
assert len(captured["input_images"]) == 16
|
||
assert captured["input_images"][-1]["image_url"] == "https://example.com/ref-15.png"
|
||
|
||
def test_rejects_non_image_local_source(self, provider, monkeypatch, tmp_path):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
text_path = tmp_path / "not-image.txt"
|
||
text_path.write_text("hello")
|
||
|
||
result = provider.generate("edit this", image_url=str(text_path))
|
||
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "invalid_image_input"
|
||
assert "not a supported image" in result["error"]
|
||
|
||
def test_rejects_svg_local_source(self, provider, monkeypatch, tmp_path):
|
||
# The shared magic-byte sniffer recognizes SVG, but gpt-image-2's
|
||
# input_image accepts raster only — SVG must fail locally with a clear
|
||
# error, not get embedded and rejected server-side with an opaque 400.
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
svg_path = tmp_path / "vector.svg"
|
||
svg_path.write_text('<svg xmlns="http://www.w3.org/2000/svg"></svg>')
|
||
|
||
result = provider.generate("edit this", image_url=str(svg_path))
|
||
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "invalid_image_input"
|
||
assert "not a supported image" in result["error"]
|
||
|
||
def test_partial_image_event_used_when_done_missing(self):
|
||
"""If output_item.done is missing, partial_image_b64 is accepted."""
|
||
payload = {
|
||
"type": "response.image_generation_call.partial_image",
|
||
"partial_image_b64": _b64_png(),
|
||
}
|
||
assert codex_plugin._extract_image_b64(payload) == _b64_png()
|
||
|
||
def test_sse_parser_handles_event_and_data_lines(self):
|
||
class _Response:
|
||
def iter_lines(self):
|
||
return iter([
|
||
"event: response.output_item.done",
|
||
'data: {"item": {"type": "image_generation_call", "result": "abc"}}',
|
||
"",
|
||
])
|
||
|
||
events = list(codex_plugin._iter_sse_json(_Response()))
|
||
assert events == [{
|
||
"type": "response.output_item.done",
|
||
"item": {"type": "image_generation_call", "result": "abc"},
|
||
}]
|
||
|
||
def test_final_response_sweep_recovers_image(self):
|
||
"""Completed response output is found by recursive payload scanning."""
|
||
payload = {
|
||
"type": "response.completed",
|
||
"response": {
|
||
"output": [{
|
||
"type": "image_generation_call",
|
||
"status": "completed",
|
||
"id": "ig_final",
|
||
"result": _b64_png(),
|
||
}],
|
||
},
|
||
}
|
||
assert codex_plugin._extract_image_b64(payload) == _b64_png()
|
||
|
||
def test_empty_response_returns_error(self, provider, monkeypatch):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
monkeypatch.setattr(codex_plugin, "_collect_image_b64", lambda *a, **kw: None)
|
||
|
||
result = provider.generate("a cat")
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "empty_response"
|
||
|
||
def test_stream_exception_returns_api_error(self, provider, monkeypatch):
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
|
||
def _boom(*args, **kwargs):
|
||
raise RuntimeError("cloudflare 403")
|
||
|
||
monkeypatch.setattr(codex_plugin, "_collect_image_b64", _boom)
|
||
|
||
result = provider.generate("a cat")
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "api_error"
|
||
assert "cloudflare 403" in result["error"]
|
||
|
||
def test_tool_choice_400_surfaces_verbatim_not_as_capability_error(
|
||
self, provider, monkeypatch
|
||
):
|
||
"""The tool_choice 400 must NOT be reported as an account limitation.
|
||
|
||
Regression for #19505 / #49008 / #31335: a previous version classified
|
||
this exact request-shape rejection as "Image generation is not enabled
|
||
for the current Codex account", telling every affected user to abandon
|
||
Codex over a bug in our own payload. The wire error must reach the user
|
||
unedited so it stays diagnosable.
|
||
|
||
Drives the REAL httpx boundary (not a mocked ``_collect_image_b64``) so
|
||
the classification path is actually exercised — mocking the collector
|
||
would skip the code under test entirely.
|
||
"""
|
||
import httpx
|
||
|
||
monkeypatch.setattr(codex_plugin, "_read_codex_access_token", lambda: "codex-token")
|
||
|
||
body = json.dumps({
|
||
"error": {
|
||
"message": "Tool choice 'image_generation' not found in 'tools' parameter.",
|
||
"type": "invalid_request_error",
|
||
"param": "tool_choice",
|
||
}
|
||
})
|
||
|
||
def _handler(request):
|
||
return httpx.Response(400, text=body, request=request)
|
||
|
||
real_client = httpx.Client
|
||
monkeypatch.setattr(
|
||
httpx,
|
||
"Client",
|
||
lambda *args, **kwargs: real_client(
|
||
transport=httpx.MockTransport(_handler),
|
||
headers=kwargs.get("headers"),
|
||
timeout=kwargs.get("timeout"),
|
||
),
|
||
)
|
||
|
||
result = provider.generate("a cat")
|
||
|
||
assert result["success"] is False
|
||
assert result["error_type"] == "api_error"
|
||
assert "HTTP 400" in result["error"]
|
||
assert "tools' parameter" in result["error"]
|
||
# The account-entitlement misdiagnosis must not come back.
|
||
assert "not enabled for the current Codex account" not in result["error"]
|
||
assert result["error_type"] != "capability_unsupported"
|
||
|
||
|
||
class TestRequestShape:
|
||
def test_payload_omits_tool_choice(self):
|
||
"""Codex rejects every tool_choice shape for hosted image_generation."""
|
||
payload = codex_plugin._build_responses_payload(
|
||
prompt="a red circle",
|
||
size="1024x1024",
|
||
quality="low",
|
||
)
|
||
assert "tool_choice" not in payload
|
||
# The hosted tool itself is still requested, and instructions do the steering.
|
||
assert payload["tools"][0]["type"] == "image_generation"
|
||
assert payload["instructions"]
|
||
|
||
def test_http_error_body_is_truncated_but_preserved(self, monkeypatch):
|
||
"""A large error body is capped at 500 chars and still surfaced."""
|
||
import httpx
|
||
|
||
body = json.dumps({
|
||
"metadata": "x" * 600,
|
||
"error": {
|
||
"message": "Tool choice 'image_generation' not found in 'tools' parameter."
|
||
},
|
||
})
|
||
|
||
def _handler(request):
|
||
return httpx.Response(400, text=body, request=request)
|
||
|
||
real_client = httpx.Client
|
||
monkeypatch.setattr(
|
||
httpx,
|
||
"Client",
|
||
lambda *args, **kwargs: real_client(
|
||
transport=httpx.MockTransport(_handler),
|
||
headers=kwargs.get("headers"),
|
||
timeout=kwargs.get("timeout"),
|
||
),
|
||
)
|
||
|
||
with pytest.raises(RuntimeError, match="HTTP 400") as excinfo:
|
||
codex_plugin._collect_image_b64(
|
||
"codex-token",
|
||
prompt="a cat",
|
||
size="1024x1024",
|
||
quality="low",
|
||
)
|
||
|
||
message = str(excinfo.value)
|
||
# Body is capped, but the actionable wire message still reaches the user.
|
||
assert "tools' parameter" in message
|
||
assert len(message) < len(body)
|
||
|
||
|
||
# ── Plugin entry point ──────────────────────────────────────────────────────
|
||
|
||
|
||
class TestRegistration:
|
||
def test_register_calls_register_image_gen_provider(self):
|
||
registered = []
|
||
|
||
class _Ctx:
|
||
def register_image_gen_provider(self, prov):
|
||
registered.append(prov)
|
||
|
||
codex_plugin.register(_Ctx())
|
||
assert len(registered) == 1
|
||
assert registered[0].name == "openai-codex"
|