Oleksandr Kozachuk
197 lines
9.6 KiB
Python
197 lines
9.6 KiB
Python
"""Unit coverage for the Responses API path (ENV-22..24, D-021)."""
|
|
|
|
import json
|
|
import unittest
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
from fjerkroa_bot.openai_responder import ENVELOPE_TEXT_FORMAT, OpenAIResponder
|
|
|
|
from .test_bdd_envelope import envelope
|
|
|
|
CONFIG = {
|
|
"openai-token": "t",
|
|
"model": "main-model",
|
|
"system": "s",
|
|
"history-limit": 5,
|
|
"use-responses-api": True,
|
|
"reasoning-effort": "medium",
|
|
}
|
|
|
|
|
|
def _msg_item():
|
|
part = Mock()
|
|
part.type = "output_text"
|
|
item = Mock()
|
|
item.type = "message"
|
|
item.content = [part]
|
|
item.model_dump = lambda: {"type": "message"}
|
|
return item
|
|
|
|
|
|
def _refusal_item():
|
|
part = Mock()
|
|
part.type = "refusal"
|
|
item = Mock()
|
|
item.type = "message"
|
|
item.content = [part]
|
|
return item
|
|
|
|
|
|
def _reasoning_item():
|
|
item = Mock()
|
|
item.type = "reasoning"
|
|
item.id = "rs_1"
|
|
item.summary = []
|
|
item.encrypted_content = "opaque-cot"
|
|
item.status = "completed" # response-only field; must NOT travel back
|
|
return item
|
|
|
|
|
|
def _call_item(name, args, call_id="call-1"):
|
|
item = Mock()
|
|
item.type = "function_call"
|
|
item.id = "fc_1"
|
|
item.name = name
|
|
item.arguments = json.dumps(args)
|
|
item.call_id = call_id
|
|
item.status = "completed"
|
|
return item
|
|
|
|
|
|
def _response(output, text=""):
|
|
result = Mock()
|
|
result.output = output
|
|
result.output_text = text
|
|
result.usage = Mock(prompt_tokens=None, completion_tokens=None, input_tokens=5, output_tokens=7)
|
|
return result
|
|
|
|
|
|
class TestResponsesPath(unittest.IsolatedAsyncioTestCase):
|
|
def _responder(self, **extra):
|
|
return OpenAIResponder(dict(CONFIG, **extra), "chat")
|
|
|
|
async def test_flag_routes_to_responses_with_reasoning(self):
|
|
"""ENV-22: flag on -> /v1/responses with envelope text.format, reasoning from config, stateless kwargs."""
|
|
responder = self._responder()
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
with patch("fjerkroa_bot.openai_responder.openai_chat", new_callable=AsyncMock) as chat_mock:
|
|
responses_mock.return_value = _response([_msg_item()], envelope(answer="hi", answer_needed=True))
|
|
answer, _ = await responder.chat([{"role": "user", "content": "hei"}], 10)
|
|
chat_mock.assert_not_awaited()
|
|
self.assertEqual(json.loads(answer["content"])["answer"], "hi")
|
|
kwargs = responses_mock.await_args.kwargs
|
|
self.assertEqual(kwargs["text"], ENVELOPE_TEXT_FORMAT)
|
|
self.assertEqual(kwargs["reasoning"], {"effort": "medium"})
|
|
self.assertFalse(kwargs["store"]) # ENV-23
|
|
self.assertIn("reasoning.encrypted_content", kwargs["include"])
|
|
|
|
async def test_flag_off_stays_on_chat_completions(self):
|
|
"""ENV-22: flag off (default) -> openai_responses never called."""
|
|
from .test_spec_structured import ok_result
|
|
|
|
responder = OpenAIResponder({k: v for k, v in CONFIG.items() if k != "use-responses-api"}, "chat")
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
with patch("fjerkroa_bot.openai_responder.openai_chat", new_callable=AsyncMock) as chat_mock:
|
|
chat_mock.return_value = ok_result()
|
|
await responder.chat([{"role": "user", "content": "hei"}], 10)
|
|
responses_mock.assert_not_awaited()
|
|
chat_mock.assert_awaited()
|
|
|
|
async def test_tools_flat_shape(self):
|
|
"""ENV-22: tools are sent in the flat Responses shape (name at top level)."""
|
|
responder = self._responder(**{"enable-news-tool": True})
|
|
responder.store = Mock() # store present -> get_news offered
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
responses_mock.return_value = _response([_msg_item()], envelope(answer="x", answer_needed=True))
|
|
await responder.chat([{"role": "user", "content": "hei"}], 10)
|
|
tools = responses_mock.await_args.kwargs["tools"]
|
|
self.assertTrue(all(tool["type"] == "function" and "name" in tool and "function" not in tool for tool in tools))
|
|
|
|
async def test_tool_loop_passes_reasoning_and_outputs_back(self):
|
|
"""ENV-23: function_call -> dispatch; next call carries reasoning item + function_call_output."""
|
|
responder = self._responder(**{"enable-news-tool": True})
|
|
responder.store = Mock()
|
|
responder._dispatch_tool = AsyncMock(return_value={"results": ["ok"]})
|
|
first = _response([_reasoning_item(), _call_item("get_news", {"topic": "x"}, "call-9")])
|
|
second = _response([_msg_item()], envelope(answer="done", answer_needed=True))
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
responses_mock.side_effect = [first, second]
|
|
answer, _ = await responder.chat([{"role": "user", "content": "news?"}], 10)
|
|
self.assertEqual(json.loads(answer["content"])["answer"], "done")
|
|
responder._dispatch_tool.assert_awaited_once()
|
|
followup_input = responses_mock.await_args_list[1].kwargs["input"]
|
|
reasoning = [item for item in followup_input if isinstance(item, dict) and item.get("type") == "reasoning"]
|
|
self.assertEqual(len(reasoning), 1)
|
|
self.assertEqual(reasoning[0]["encrypted_content"], "opaque-cot")
|
|
self.assertNotIn("status", reasoning[0]) # response-only field stripped (live-400 regression)
|
|
calls_back = [item for item in followup_input if isinstance(item, dict) and item.get("type") == "function_call"]
|
|
self.assertNotIn("status", calls_back[0])
|
|
outputs = [item for item in followup_input if isinstance(item, dict) and item.get("type") == "function_call_output"]
|
|
self.assertEqual(len(outputs), 1)
|
|
self.assertEqual(outputs[0]["call_id"], "call-9")
|
|
|
|
async def test_exhausted_rounds_force_toolless_answer(self):
|
|
"""ENV-23: after responses-tool-rounds rounds the final call drops tools."""
|
|
responder = self._responder(**{"enable-news-tool": True, "responses-tool-rounds": 1})
|
|
responder.store = Mock()
|
|
responder._dispatch_tool = AsyncMock(return_value={"results": []})
|
|
looping = _response([_call_item("get_news", {}, "c")])
|
|
final = _response([_msg_item()], envelope(answer="forced", answer_needed=True))
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
responses_mock.side_effect = [looping, final]
|
|
answer, _ = await responder.chat([{"role": "user", "content": "go"}], 10)
|
|
self.assertEqual(json.loads(answer["content"])["answer"], "forced")
|
|
self.assertNotIn("tools", responses_mock.await_args_list[1].kwargs)
|
|
|
|
async def test_refusal_is_failed_attempt(self):
|
|
"""ENV-24: a refusal part -> no answer."""
|
|
responder = self._responder()
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
responses_mock.return_value = _response([_refusal_item()])
|
|
answer, _ = await responder.chat([{"role": "user", "content": "hei"}], 10)
|
|
self.assertIsNone(answer)
|
|
|
|
async def test_vision_parts_mapped(self):
|
|
"""ENV-22: chat-format image parts become input_image items."""
|
|
items = OpenAIResponder._responses_input(
|
|
[
|
|
{"role": "user", "content": [{"type": "text", "text": "look"}, {"type": "image_url", "image_url": {"url": "data:x"}}]},
|
|
{"role": "tool", "content": "dropped"},
|
|
{"role": "assistant", "content": "{}"},
|
|
]
|
|
)
|
|
self.assertEqual(items[0]["content"][0], {"type": "input_text", "text": "look"})
|
|
self.assertEqual(items[0]["content"][1], {"type": "input_image", "image_url": "data:x"})
|
|
self.assertEqual(len(items), 2) # tool row dropped
|
|
|
|
|
|
class TestToolVisionInjection(unittest.IsolatedAsyncioTestCase):
|
|
def _responder(self, **extra):
|
|
return OpenAIResponder(dict(CONFIG, **extra), "chat")
|
|
|
|
async def test_tool_vision_images_attached_as_input_image(self):
|
|
"""URL-09: a tool result's vision data URLs become input_image items; never JSON text."""
|
|
responder = self._responder(**{"enable-news-tool": True})
|
|
responder.store = Mock()
|
|
responder._dispatch_tool = AsyncMock(
|
|
return_value={"url": "u", "text": "t", "images_cached": 1, "vision": ["data:image/png;base64,AAA"]}
|
|
)
|
|
first = _response([_call_item("fetch_url", {"url": "https://xkcd.com/1"}, "call-2")])
|
|
second = _response([_msg_item()], envelope(answer="seen", answer_needed=True))
|
|
with patch("fjerkroa_bot.openai_responder.openai_responses", new_callable=AsyncMock) as responses_mock:
|
|
responses_mock.side_effect = [first, second]
|
|
answer, _ = await responder.chat([{"role": "user", "content": "look at this"}], 10)
|
|
self.assertEqual(json.loads(answer["content"])["answer"], "seen")
|
|
followup = responses_mock.await_args_list[1].kwargs["input"]
|
|
image_parts = [
|
|
part
|
|
for item in followup
|
|
if isinstance(item, dict) and isinstance(item.get("content"), list)
|
|
for part in item["content"]
|
|
if part.get("type") == "input_image"
|
|
]
|
|
self.assertEqual(image_parts[0]["image_url"], "data:image/png;base64,AAA")
|
|
outputs = [item for item in followup if isinstance(item, dict) and item.get("type") == "function_call_output"]
|
|
self.assertNotIn("data:image", outputs[0]["output"]) # data URL never in JSON tool text
|
|
self.assertNotIn("vision", outputs[0]["output"])
|