mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-07-09 13:21:42 +00:00
019950560d
Replace the plugin-local _IMAGE_MAGIC_MIME table + _sniff_image_mime body with a delegation to agent.image_routing._sniff_mime_from_bytes, the canonical magic-byte sniffer already used across the codebase, then gate its result to the raster formats gpt-image-2's Responses input_image actually accepts (png/jpeg/gif/webp). The shared sniffer also recognizes SVG/TIFF/ICO; without the allowlist those would pass local validation and be rejected server-side with an opaque HTTP 400. Gating locally fails them cleanly as invalid_image_input. Adds a regression test for SVG rejection. Follow-up on top of @CrazyBoyM's #55828.
324 lines
13 KiB
Python
324 lines
13 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
|
||
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"
|
||
assert captured["tool_choice"]["type"] == "allowed_tools"
|
||
assert captured["tool_choice"]["mode"] == "required"
|
||
assert captured["tool_choice"]["tools"] == [{"type": "image_generation"}]
|
||
|
||
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"]
|
||
|
||
|
||
# ── 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"
|