Oleksandr Kozachuk
61 lines
3.3 KiB
Python
61 lines
3.3 KiB
Python
"""Unit coverage for FDB-005 structured outputs (ENV-18, ENV-19)."""
|
|
|
|
import unittest
|
|
from unittest.mock import AsyncMock, Mock, patch
|
|
|
|
from fjerkroa_bot.ai_responder import AIMessage
|
|
from fjerkroa_bot.openai_responder import ENVELOPE_RESPONSE_FORMAT, OpenAIResponder
|
|
|
|
from .test_bdd_envelope import FakeModelResponder, envelope
|
|
|
|
RESPONDER_CONFIG = {"openai-token": "test", "model": "main-model", "system": "s", "history-limit": 5}
|
|
|
|
|
|
def ok_result(content=None):
|
|
message = Mock(content=content or envelope(answer="x", answer_needed=True), role="assistant", tool_calls=None, refusal=None)
|
|
return Mock(choices=[Mock(message=message)], usage="usage")
|
|
|
|
|
|
class TestNoRepairPath(unittest.IsolatedAsyncioTestCase):
|
|
async def test_malformed_output_is_failed_attempt(self):
|
|
"""ENV-18: malformed output is a failed attempt with backoff — no repair path (replaces ENV-06)."""
|
|
responder = FakeModelResponder({"system": "s", "history-limit": 5}, "chat")
|
|
responder.scripted = ["definitely not json", "definitely not json", "definitely not json"]
|
|
with patch("fjerkroa_bot.ai_responder.asyncio.sleep", new_callable=AsyncMock) as sleep:
|
|
with self.assertRaises(RuntimeError):
|
|
await responder.send(AIMessage("alice", "hei", "chat"))
|
|
self.assertEqual(responder.chat_calls, 3)
|
|
self.assertGreaterEqual(sleep.await_count, 2)
|
|
self.assertFalse(hasattr(responder, "fix"))
|
|
|
|
async def test_refusal_is_failed_attempt(self):
|
|
"""ENV-18: a model refusal yields no answer from chat()."""
|
|
responder = OpenAIResponder(RESPONDER_CONFIG, "chat")
|
|
refusal_message = Mock(content=None, refusal="I cannot help with that.", tool_calls=None, role="assistant")
|
|
with patch("fjerkroa_bot.openai_responder.openai_chat", new_callable=AsyncMock) as chat_mock:
|
|
chat_mock.return_value = Mock(choices=[Mock(message=refusal_message)], usage="usage")
|
|
answer, _ = await responder.chat([{"role": "user", "content": "hi"}], 10)
|
|
self.assertIsNone(answer)
|
|
|
|
|
|
class TestEnvelopeSchema(unittest.IsolatedAsyncioTestCase):
|
|
def test_schema_shape_pinned(self):
|
|
"""ENV-19: strict envelope schema — exact fields, all required, closed object."""
|
|
json_schema = ENVELOPE_RESPONSE_FORMAT["json_schema"]
|
|
schema = json_schema["schema"]
|
|
expected = {"answer", "answer_needed", "channel", "staff", "picture", "picture_count", "picture_edit", "hack"}
|
|
self.assertEqual(set(schema["properties"]), expected)
|
|
self.assertEqual(set(schema["required"]), expected)
|
|
self.assertFalse(schema["additionalProperties"])
|
|
self.assertTrue(json_schema["strict"])
|
|
self.assertEqual(json_schema["name"], "envelope")
|
|
|
|
async def test_chat_carries_response_format(self):
|
|
"""ENV-19: chat calls pass the pinned response_format to the API."""
|
|
responder = OpenAIResponder(RESPONDER_CONFIG, "chat")
|
|
with patch("fjerkroa_bot.openai_responder.openai_chat", new_callable=AsyncMock) as chat_mock:
|
|
chat_mock.return_value = ok_result()
|
|
answer, _ = await responder.chat([{"role": "user", "content": "hi"}], 10)
|
|
self.assertIsNotNone(answer)
|
|
self.assertEqual(chat_mock.await_args.kwargs["response_format"], ENVELOPE_RESPONSE_FORMAT)
|