File size: 6,295 Bytes
108e227 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """End-to-end tests for inline image inputs on /v1/chat/completions and /v1/responses.
Covers the multimodal normalization path added to the API server. Unlike the
adapter-level tests that patch ``_run_agent``, these tests patch
``AIAgent.run_conversation`` instead so the adapter's full request-handling
path (including the ``run_agent`` prologue that used to crash on list content)
executes against a real aiohttp app.
"""
from unittest.mock import MagicMock, patch
import pytest
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
from gateway.config import PlatformConfig
from gateway.platforms.api_server import (
APIServerAdapter,
_content_has_visible_payload,
_normalize_multimodal_content,
cors_middleware,
security_headers_middleware,
)
# ---------------------------------------------------------------------------
# Pure-function tests for _normalize_multimodal_content
# ---------------------------------------------------------------------------
class TestNormalizeMultimodalContent:
def test_string_passthrough(self):
assert _normalize_multimodal_content("hello") == "hello"
def test_none_returns_empty_string(self):
assert _normalize_multimodal_content(None) == ""
def test_text_only_list_collapses_to_string(self):
content = [{"type": "text", "text": "hi"}, {"type": "text", "text": "there"}]
assert _normalize_multimodal_content(content) == "hi\nthere"
def test_responses_input_text_canonicalized(self):
content = [{"type": "input_text", "text": "hello"}]
assert _normalize_multimodal_content(content) == "hello"
def test_input_image_converted_to_canonical_shape(self):
content = [
{"type": "input_text", "text": "hi"},
{"type": "input_image", "image_url": "https://example.com/cat.png"},
]
out = _normalize_multimodal_content(content)
assert out == [
{"type": "text", "text": "hi"},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
]
class TestContentHasVisiblePayload:
def test_list_with_image_only(self):
assert _content_has_visible_payload([{"type": "image_url", "image_url": {"url": "x"}}])
# ---------------------------------------------------------------------------
# HTTP integration — real aiohttp client hitting the adapter handlers
# ---------------------------------------------------------------------------
def _make_adapter() -> APIServerAdapter:
return APIServerAdapter(PlatformConfig(enabled=True))
def _create_app(adapter: APIServerAdapter) -> web.Application:
mws = [mw for mw in (cors_middleware, security_headers_middleware) if mw is not None]
app = web.Application(middlewares=mws)
app["api_server_adapter"] = adapter
app.router.add_post("/v1/chat/completions", adapter._handle_chat_completions)
app.router.add_post("/v1/responses", adapter._handle_responses)
app.router.add_get("/v1/responses/{response_id}", adapter._handle_get_response)
return app
@pytest.fixture
def adapter():
return _make_adapter()
class TestChatCompletionsMultimodalHTTP:
@pytest.mark.asyncio
async def test_inline_image_preserved_to_run_agent(self, adapter):
"""Multimodal user content reaches _run_agent as a list of parts."""
image_payload = [
{"type": "text", "text": "What's in this image?"},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png", "detail": "high"}},
]
app = _create_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(
adapter,
"_run_agent",
new=MagicMock(),
) as mock_run:
async def _stub(**kwargs):
mock_run.captured = kwargs
return (
{"final_response": "A cat.", "messages": [], "api_calls": 1},
{"input_tokens": 0, "output_tokens": 0, "total_tokens": 0},
)
mock_run.side_effect = _stub
resp = await cli.post(
"/v1/chat/completions",
json={
"model": "hermes-agent",
"messages": [{"role": "user", "content": image_payload}],
},
)
assert resp.status == 200, await resp.text()
assert mock_run.captured["user_message"] == image_payload
class TestResponsesMultimodalHTTP:
@pytest.mark.asyncio
async def test_input_image_canonicalized_and_forwarded(self, adapter):
app = _create_app(adapter)
async with TestClient(TestServer(app)) as cli:
with patch.object(adapter, "_run_agent", new=MagicMock()) as mock_run:
async def _stub(**kwargs):
mock_run.captured = kwargs
return (
{"final_response": "ok", "messages": [], "api_calls": 1},
{"input_tokens": 0, "output_tokens": 0, "total_tokens": 0},
)
mock_run.side_effect = _stub
resp = await cli.post(
"/v1/responses",
json={
"model": "hermes-agent",
"input": [
{
"role": "user",
"content": [
{"type": "input_text", "text": "Describe."},
{
"type": "input_image",
"image_url": "https://example.com/cat.png",
},
],
}
],
},
)
assert resp.status == 200, await resp.text()
expected = [
{"type": "text", "text": "Describe."},
{"type": "image_url", "image_url": {"url": "https://example.com/cat.png"}},
]
assert mock_run.captured["user_message"] == expected
|