Eduardo David Paredes Vara
82 lines
4.0 KiB
Python
82 lines
4.0 KiB
Python
"""Run inside the LiteLLM runtime: python < test_codex_provider.py."""
|
|
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from codex_provider import _accumulate_tool_chunk, _event_chunk, _split_messages
|
|
from codex_provider import CodexAudioAuth
|
|
|
|
|
|
class CodexAudioAuthTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_resolves_credential_only_for_opted_in_audio_endpoint(self):
|
|
hook = CodexAudioAuth()
|
|
untouched = {"model": "gpt-6.1-sol", "api_key": "other"}
|
|
with patch("codex_provider._get_auth") as auth:
|
|
self.assertIsNone(await hook.async_pre_call_deployment_hook(untouched))
|
|
auth.assert_not_called()
|
|
for endpoint in ("https://untrusted.example/v1", "https://api.openai.com/v1"):
|
|
with self.assertRaises(ValueError):
|
|
await hook.async_pre_call_deployment_hook({
|
|
"api_base": endpoint, "model": "unrelated-model",
|
|
"api_key": "codex-oauth-runtime"})
|
|
request = {"api_base": "https://api.openai.com/v1",
|
|
"model": "openai/gpt-4o-mini-transcribe",
|
|
"api_key": "codex-oauth-runtime"}
|
|
with patch("codex_provider._get_auth", return_value={"tokens": {
|
|
"access_token": "test-token", "account_id": "test-account"}}):
|
|
result = await hook.async_pre_call_deployment_hook(request)
|
|
self.assertEqual(result["api_key"], "test-token")
|
|
self.assertEqual(result["extra_headers"]["ChatGPT-Account-Id"], "test-account")
|
|
|
|
|
|
class CodexStreamingTests(unittest.TestCase):
|
|
def test_parallel_calls_keep_complete_json_and_dense_indices(self):
|
|
state, calls = {}, {}
|
|
items = [
|
|
{"type": "function_call", "call_id": "call_a", "name": "fetch",
|
|
"arguments": '{"url":"https://example.com/a"}'},
|
|
{"type": "function_call", "call_id": "call_b", "name": "fetch",
|
|
"arguments": '{"url":"https://example.com/b"}'},
|
|
]
|
|
# Responses output indices include reasoning/text; tool indices must not.
|
|
for output_index, item in zip((1, 3), items):
|
|
self.assertIsNone(_event_chunk({
|
|
"type": "response.output_item.added", "output_index": output_index,
|
|
"item": {**item, "arguments": ""},
|
|
}, state))
|
|
for output_index, item in zip((3, 1), reversed(items)):
|
|
self.assertIsNone(_event_chunk({
|
|
"type": "response.function_call_arguments.delta",
|
|
"output_index": output_index, "delta": item["arguments"],
|
|
}, state))
|
|
event = {"type": "response.output_item.done",
|
|
"output_index": output_index, "item": item}
|
|
chunk = _event_chunk(event, state)
|
|
_accumulate_tool_chunk(calls, chunk)
|
|
self.assertIsNone(_event_chunk(event, state))
|
|
self.assertEqual(sorted(calls), [0, 1])
|
|
for index, item in enumerate(items):
|
|
self.assertEqual(calls[index]["id"], item["call_id"])
|
|
self.assertEqual(calls[index]["function"], {
|
|
"name": item["name"], "arguments": item["arguments"],
|
|
})
|
|
finish = _event_chunk({"type": "response.completed"}, state)
|
|
self.assertEqual(finish["finish_reason"], "tool_calls")
|
|
|
|
def test_text_stays_streamed(self):
|
|
chunk = _event_chunk({"type": "response.output_text.delta", "delta": "OK"}, {})
|
|
self.assertEqual(chunk["text"], "OK")
|
|
self.assertFalse(chunk["is_finished"])
|
|
|
|
def test_chat_content_blocks_are_converted_without_mutating_input(self):
|
|
messages = [{"role": "system", "content": [{"type": "text", "text": "Help"}]},
|
|
{"role": "user", "content": [{"type": "text", "text": "Hi"}]}]
|
|
instructions, items = _split_messages(messages)
|
|
self.assertEqual(instructions, "Help")
|
|
self.assertEqual(items[0]["content"], [{"type": "input_text", "text": "Hi"}])
|
|
self.assertEqual(messages[1]["content"][0]["type"], "text")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|