284 lines
10 KiB
Python
284 lines
10 KiB
Python
"""Round-trip test for the OpenAI Responses bridge route (PR2).
|
|
|
|
Drives the bridge over a raw aiohttp client to stay independent of any
|
|
specific OpenAI SDK version. Covers JSON mode + SSE streaming, reasoning,
|
|
function_call output rendering, and the 401 unknown-token path.
|
|
|
|
The strict event-sequence properties of the SSE stream are covered by
|
|
``test_responses_event_sequence`` (which tests the synthesizer directly,
|
|
no HTTP). This file exercises the full handler chain end-to-end.
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import pytest
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
from evalscope.agent.external.bridge import ModelProxyServer
|
|
from evalscope.api.messages import ChatMessageAssistant, ContentReasoning, ContentText
|
|
from evalscope.api.model import GenerateConfig, Model, ModelOutput
|
|
from evalscope.api.model.model_output import ChatCompletionChoice
|
|
from evalscope.api.tool import ToolCall, ToolFunction
|
|
from evalscope.models.mockllm import MockLLM
|
|
from evalscope.utils.function_utils import AsyncioLoopRunner
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _release_bridge_loop():
|
|
yield
|
|
AsyncioLoopRunner.shutdown_for_thread()
|
|
|
|
|
|
def _build_model(output: ModelOutput) -> Model:
|
|
api = MockLLM(model_name='mock-responses', custom_outputs=[output])
|
|
return Model(api=api, config=GenerateConfig())
|
|
|
|
|
|
def _text_output(text: str) -> ModelOutput:
|
|
msg = ChatMessageAssistant(content=text)
|
|
return ModelOutput(model='mock-responses', choices=[ChatCompletionChoice(message=msg, stop_reason='stop')])
|
|
|
|
|
|
def _reasoning_output(reasoning: str, text: str) -> ModelOutput:
|
|
msg = ChatMessageAssistant(content=[ContentReasoning(reasoning=reasoning), ContentText(text=text)])
|
|
return ModelOutput(model='mock-responses', choices=[ChatCompletionChoice(message=msg, stop_reason='stop')])
|
|
|
|
|
|
def _function_call_output(name: str, args: dict, call_id: str = 'call-abc') -> ModelOutput:
|
|
tc = ToolCall(id=call_id, function=ToolFunction(name=name, arguments=args), type='function')
|
|
msg = ChatMessageAssistant(content='', tool_calls=[tc])
|
|
return ModelOutput(model='mock-responses', choices=[ChatCompletionChoice(message=msg, stop_reason='tool_calls')])
|
|
|
|
|
|
def _user_input_message(text: str) -> dict:
|
|
return {'type': 'message', 'role': 'user', 'content': [{'type': 'input_text', 'text': text}]}
|
|
|
|
|
|
def _post_json(url: str, body: dict, token: str) -> dict:
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(body).encode('utf-8'),
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'Bearer {token}',
|
|
},
|
|
method='POST',
|
|
)
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return json.loads(resp.read().decode('utf-8'))
|
|
|
|
|
|
def _post_stream(url: str, body: dict, token: str) -> str:
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=json.dumps(body).encode('utf-8'),
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': f'Bearer {token}',
|
|
'Accept': 'text/event-stream',
|
|
},
|
|
method='POST',
|
|
)
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
return resp.read().decode('utf-8')
|
|
|
|
|
|
def _parse_responses_sse(raw: str) -> list:
|
|
"""Parse ``event: <type>\\ndata: <json>\\n\\n`` frames into ``(type, data)`` pairs."""
|
|
out = []
|
|
for chunk in raw.split('\n\n'):
|
|
lines = chunk.strip().split('\n')
|
|
if len(lines) < 2 or not lines[0].startswith('event: '):
|
|
continue
|
|
event = lines[0][len('event: '):]
|
|
data_line = next((ln for ln in lines if ln.startswith('data: ')), None)
|
|
if data_line is None:
|
|
continue
|
|
out.append((event, json.loads(data_line[len('data: '):])))
|
|
return out
|
|
|
|
|
|
def test_responses_json_message_payload_shape():
|
|
"""JSON-mode response has correct top-level + output[message] item shape."""
|
|
|
|
async def _go():
|
|
proxy = await ModelProxyServer.get_or_start()
|
|
model = _build_model(_text_output('hello from responses'))
|
|
async with proxy.trial_session(model=model, framework='mock') as session:
|
|
url = f'{proxy.base_url}/openai/v1/responses'
|
|
|
|
def _request():
|
|
return _post_json(
|
|
url,
|
|
{
|
|
'model': 'mock-responses',
|
|
'input': [_user_input_message('hi')],
|
|
},
|
|
session.token,
|
|
)
|
|
|
|
return await asyncio.get_running_loop().run_in_executor(None, _request)
|
|
|
|
body = AsyncioLoopRunner.run(_go())
|
|
assert body['object'] == 'response'
|
|
assert body['status'] == 'completed'
|
|
assert body['model'] == 'mock-responses'
|
|
assert len(body['output']) == 1
|
|
item = body['output'][0]
|
|
assert item['type'] == 'message'
|
|
assert item['role'] == 'assistant'
|
|
assert item['content'][0]['type'] == 'output_text'
|
|
assert item['content'][0]['text'] == 'hello from responses'
|
|
assert set(body['usage']) == {'input_tokens', 'output_tokens', 'total_tokens'}
|
|
|
|
|
|
def test_responses_json_reasoning_renders_summary_item():
|
|
"""ContentReasoning block → output item with type=reasoning, summary_text."""
|
|
|
|
async def _go():
|
|
proxy = await ModelProxyServer.get_or_start()
|
|
model = _build_model(_reasoning_output('thinking step by step', '42'))
|
|
async with proxy.trial_session(model=model, framework='mock') as session:
|
|
url = f'{proxy.base_url}/openai/v1/responses'
|
|
|
|
def _request():
|
|
return _post_json(
|
|
url,
|
|
{
|
|
'model': 'mock-responses',
|
|
'input': [_user_input_message('why?')],
|
|
},
|
|
session.token,
|
|
)
|
|
|
|
return await asyncio.get_running_loop().run_in_executor(None, _request)
|
|
|
|
body = AsyncioLoopRunner.run(_go())
|
|
types = [item['type'] for item in body['output']]
|
|
assert types == ['reasoning', 'message']
|
|
reasoning_item = body['output'][0]
|
|
assert reasoning_item['summary'][0]['type'] == 'summary_text'
|
|
assert reasoning_item['summary'][0]['text'] == 'thinking step by step'
|
|
msg_item = body['output'][1]
|
|
assert msg_item['content'][0]['text'] == '42'
|
|
|
|
|
|
def test_responses_json_function_call_uses_string_arguments():
|
|
"""function_call output item: arguments is a JSON-encoded string (per spec)."""
|
|
|
|
async def _go():
|
|
proxy = await ModelProxyServer.get_or_start()
|
|
model = _build_model(_function_call_output('lookup', {'q': 'cats', 'limit': 3}))
|
|
async with proxy.trial_session(model=model, framework='mock') as session:
|
|
url = f'{proxy.base_url}/openai/v1/responses'
|
|
|
|
def _request():
|
|
return _post_json(
|
|
url,
|
|
{
|
|
'model': 'mock-responses',
|
|
'input': [_user_input_message('find me cats')],
|
|
'tools': [{
|
|
'type': 'function',
|
|
'name': 'lookup',
|
|
'description': 'find things',
|
|
'parameters': {
|
|
'type': 'object',
|
|
'properties': {
|
|
'q': {'type': 'string'},
|
|
'limit': {'type': 'integer'},
|
|
},
|
|
'required': ['q'],
|
|
},
|
|
}],
|
|
},
|
|
session.token,
|
|
)
|
|
|
|
return await asyncio.get_running_loop().run_in_executor(None, _request)
|
|
|
|
body = AsyncioLoopRunner.run(_go())
|
|
fc_items = [it for it in body['output'] if it['type'] == 'function_call']
|
|
assert len(fc_items) == 1
|
|
fc = fc_items[0]
|
|
assert fc['name'] == 'lookup'
|
|
assert fc['call_id'] == 'call-abc'
|
|
# arguments MUST be JSON-encoded string per Responses spec
|
|
assert isinstance(fc['arguments'], str)
|
|
assert json.loads(fc['arguments']) == {'q': 'cats', 'limit': 3}
|
|
|
|
|
|
def test_responses_streaming_text_round_trip_reassembles_via_output_text_deltas():
|
|
"""SSE deltas reassemble to the original text; required events present."""
|
|
expected = 'streamed responses output: 42 ' * 4
|
|
|
|
async def _go():
|
|
proxy = await ModelProxyServer.get_or_start()
|
|
model = _build_model(_text_output(expected))
|
|
async with proxy.trial_session(model=model, framework='mock') as session:
|
|
url = f'{proxy.base_url}/openai/v1/responses'
|
|
|
|
def _request():
|
|
return _post_stream(
|
|
url,
|
|
{
|
|
'model': 'mock-responses',
|
|
'input': [_user_input_message('stream?')],
|
|
'stream': True,
|
|
},
|
|
session.token,
|
|
)
|
|
|
|
return await asyncio.get_running_loop().run_in_executor(None, _request)
|
|
|
|
raw = AsyncioLoopRunner.run(_go())
|
|
events = _parse_responses_sse(raw)
|
|
event_types = [t for t, _ in events]
|
|
|
|
# Spec bookends.
|
|
assert event_types[0] == 'response.created'
|
|
assert event_types[1] == 'response.in_progress'
|
|
assert event_types[-1] == 'response.completed'
|
|
# No [DONE] sentinel.
|
|
assert '[DONE]' not in raw
|
|
|
|
# Reassemble text from output_text.delta frames.
|
|
reassembled = ''.join(
|
|
data['delta'] for ev, data in events if ev == 'response.output_text.delta'
|
|
)
|
|
assert reassembled == expected
|
|
|
|
# sequence_number is strictly monotonic.
|
|
seqs = [data['sequence_number'] for _, data in events]
|
|
assert seqs == list(range(1, len(seqs) + 1))
|
|
|
|
|
|
def test_responses_unknown_token_returns_401():
|
|
"""The bridge must 401 a Responses request with an unknown trial token."""
|
|
|
|
async def _go() -> int:
|
|
proxy = await ModelProxyServer.get_or_start()
|
|
url = f'{proxy.base_url}/openai/v1/responses'
|
|
|
|
def _request() -> int:
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=b'{"model":"x","input":[]}',
|
|
headers={
|
|
'Content-Type': 'application/json',
|
|
'Authorization': 'Bearer trial-bogus',
|
|
},
|
|
method='POST',
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=5) as resp:
|
|
return resp.status
|
|
except urllib.error.HTTPError as exc:
|
|
return exc.code
|
|
|
|
return await asyncio.get_running_loop().run_in_executor(None, _request)
|
|
|
|
status = AsyncioLoopRunner.run(_go())
|
|
assert status == 401
|