wehub-resource-sync
296 lines
10 KiB
Python
296 lines
10 KiB
Python
from typing import cast
|
|
|
|
import pytest
|
|
from openai.types.realtime.realtime_conversation_item_user_message import (
|
|
RealtimeConversationItemUserMessage,
|
|
)
|
|
from openai.types.realtime.realtime_response_usage import RealtimeResponseUsage
|
|
from openai.types.realtime.realtime_tracing_config import (
|
|
TracingConfiguration,
|
|
)
|
|
|
|
from agents import Agent, function_tool, tool_namespace
|
|
from agents.exceptions import UserError
|
|
from agents.handoffs import handoff
|
|
from agents.realtime.config import RealtimeModelTracingConfig
|
|
from agents.realtime.model_inputs import (
|
|
RealtimeModelSendRawMessage,
|
|
RealtimeModelSendUserInput,
|
|
RealtimeModelUserInputMessage,
|
|
)
|
|
from agents.realtime.openai_realtime import (
|
|
OpenAIRealtimeWebSocketModel,
|
|
_ConversionHelper,
|
|
get_api_key,
|
|
)
|
|
from agents.tool import Tool
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_api_key_from_env(monkeypatch):
|
|
monkeypatch.setenv("OPENAI_API_KEY", "env-key")
|
|
assert await get_api_key(None) == "env-key"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_api_key_from_callable_async():
|
|
async def f():
|
|
return "k"
|
|
|
|
assert await get_api_key(f) == "k"
|
|
|
|
|
|
def test_try_convert_raw_message_invalid_returns_none():
|
|
msg = RealtimeModelSendRawMessage(message={"type": "invalid.event", "other_data": {}})
|
|
assert _ConversionHelper.try_convert_raw_message(msg) is None
|
|
|
|
|
|
def test_convert_response_usage_preserves_modality_details():
|
|
event = _ConversionHelper.convert_response_usage(
|
|
RealtimeResponseUsage.model_validate(
|
|
{
|
|
"total_tokens": 253,
|
|
"input_tokens": 132,
|
|
"output_tokens": 121,
|
|
"input_token_details": {
|
|
"text_tokens": 119,
|
|
"audio_tokens": 13,
|
|
"image_tokens": 0,
|
|
"cached_tokens": 64,
|
|
"cached_tokens_details": {
|
|
"text_tokens": 60,
|
|
"audio_tokens": 4,
|
|
"image_tokens": 0,
|
|
},
|
|
},
|
|
"output_token_details": {"text_tokens": 30, "audio_tokens": 91},
|
|
}
|
|
)
|
|
)
|
|
|
|
assert event.usage.requests == 1
|
|
assert event.usage.input_tokens == 132
|
|
assert event.usage.output_tokens == 121
|
|
assert event.usage.total_tokens == 253
|
|
assert event.usage.input_tokens_details.cached_tokens == 64
|
|
assert event.input_tokens_details is not None
|
|
assert event.input_tokens_details.text_tokens == 119
|
|
assert event.input_tokens_details.audio_tokens == 13
|
|
assert event.input_tokens_details.image_tokens == 0
|
|
assert event.input_tokens_details.cached_tokens == 64
|
|
assert event.input_tokens_details.cached_tokens_details is not None
|
|
assert event.input_tokens_details.cached_tokens_details.text_tokens == 60
|
|
assert event.input_tokens_details.cached_tokens_details.audio_tokens == 4
|
|
assert event.input_tokens_details.cached_tokens_details.image_tokens == 0
|
|
assert event.output_tokens_details is not None
|
|
assert event.output_tokens_details.text_tokens == 30
|
|
assert event.output_tokens_details.audio_tokens == 91
|
|
|
|
|
|
def test_convert_response_usage_preserves_missing_details_and_derives_total():
|
|
event = _ConversionHelper.convert_response_usage(
|
|
RealtimeResponseUsage.model_validate(
|
|
{
|
|
"input_tokens": 12,
|
|
"output_tokens": 3,
|
|
"input_token_details": {"audio_tokens": 0},
|
|
}
|
|
)
|
|
)
|
|
|
|
assert event.usage.total_tokens == 15
|
|
assert event.input_tokens_details is not None
|
|
assert event.input_tokens_details.audio_tokens == 0
|
|
assert event.input_tokens_details.text_tokens is None
|
|
assert event.input_tokens_details.cached_tokens is None
|
|
assert event.input_tokens_details.cached_tokens_details is None
|
|
assert event.output_tokens_details is None
|
|
|
|
|
|
def test_convert_user_input_to_conversation_item_dict_and_str():
|
|
# Dict with mixed, including unknown parts (silently skipped)
|
|
dict_input_any = {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "input_text", "text": "hello"},
|
|
{"type": "input_image", "image_url": "http://x/y.png", "detail": "auto"},
|
|
{"type": "bogus", "x": 1},
|
|
],
|
|
}
|
|
event = RealtimeModelSendUserInput(
|
|
user_input=cast(RealtimeModelUserInputMessage, dict_input_any)
|
|
)
|
|
item_any = _ConversionHelper.convert_user_input_to_conversation_item(event)
|
|
item = cast(RealtimeConversationItemUserMessage, item_any)
|
|
assert item.role == "user"
|
|
|
|
# String input becomes input_text
|
|
event2 = RealtimeModelSendUserInput(user_input="hi")
|
|
item2_any = _ConversionHelper.convert_user_input_to_conversation_item(event2)
|
|
item2 = cast(RealtimeConversationItemUserMessage, item2_any)
|
|
assert item2.content[0].type == "input_text"
|
|
|
|
|
|
def test_convert_user_input_dict_skips_invalid_input_text_parts():
|
|
"""input_text parts with missing/non-string text must be skipped, not
|
|
forwarded as Content(text=None) which the realtime API rejects."""
|
|
dict_input_any = {
|
|
"type": "message",
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "input_text"}, # missing text
|
|
{"type": "input_text", "text": 123}, # non-string text
|
|
{"type": "input_text", "text": "ok"}, # valid
|
|
],
|
|
}
|
|
event = RealtimeModelSendUserInput(
|
|
user_input=cast(RealtimeModelUserInputMessage, dict_input_any)
|
|
)
|
|
item = cast(
|
|
RealtimeConversationItemUserMessage,
|
|
_ConversionHelper.convert_user_input_to_conversation_item(event),
|
|
)
|
|
assert item.content is not None
|
|
assert len(item.content) == 1
|
|
assert item.content[0].type == "input_text"
|
|
assert item.content[0].text == "ok"
|
|
|
|
|
|
def test_convert_tracing_config_variants():
|
|
from agents.realtime.openai_realtime import _ConversionHelper as CH
|
|
|
|
assert CH.convert_tracing_config(None) is None
|
|
assert CH.convert_tracing_config("auto") == "auto"
|
|
cfg: RealtimeModelTracingConfig = {
|
|
"group_id": "g",
|
|
"metadata": {"k": "v"},
|
|
"workflow_name": "wf",
|
|
}
|
|
oc_any = CH.convert_tracing_config(cfg)
|
|
oc = cast(TracingConfiguration, oc_any)
|
|
assert oc.group_id == "g"
|
|
assert oc.workflow_name == "wf"
|
|
|
|
|
|
def test_tools_to_session_tools_raises_on_non_function_tool():
|
|
class NotFunctionTool:
|
|
def __init__(self):
|
|
self.name = "x"
|
|
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
with pytest.raises(UserError):
|
|
m._tools_to_session_tools(cast(list[Tool], [NotFunctionTool()]), [])
|
|
|
|
|
|
def test_tools_to_session_tools_includes_handoffs():
|
|
a = Agent(name="a")
|
|
h = handoff(a)
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
out = m._tools_to_session_tools([], [h])
|
|
assert out[0].name is not None and out[0].name.startswith("transfer_to_")
|
|
|
|
|
|
def test_tools_to_session_tools_rejects_duplicate_function_tool_names():
|
|
tool_one = function_tool(lambda: "one", name_override="lookup_account")
|
|
tool_two = function_tool(lambda: "two", name_override="lookup_account")
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
with pytest.raises(
|
|
UserError,
|
|
match=("Duplicate Realtime tool name found: 'lookup_account' \\(2 function tools\\)"),
|
|
):
|
|
m._tools_to_session_tools([tool_one, tool_two], [])
|
|
|
|
|
|
def test_tools_to_session_tools_rejects_function_handoff_name_conflict():
|
|
tool = function_tool(lambda: "ok", name_override="transfer_to_billing")
|
|
h = handoff(Agent(name="billing"), tool_name_override="transfer_to_billing")
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
with pytest.raises(
|
|
UserError,
|
|
match=(
|
|
"Duplicate Realtime tool name found: "
|
|
"'transfer_to_billing' \\(function tool and handoff\\)"
|
|
),
|
|
):
|
|
m._tools_to_session_tools([tool], [h])
|
|
|
|
|
|
def test_tools_to_session_tools_ignores_disabled_function_tool_name_conflict():
|
|
tool = function_tool(
|
|
lambda: "ok",
|
|
name_override="transfer_to_billing",
|
|
is_enabled=False,
|
|
)
|
|
h = handoff(Agent(name="billing"), tool_name_override="transfer_to_billing")
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
out = m._tools_to_session_tools([tool], [h])
|
|
|
|
assert [tool.name for tool in out] == ["transfer_to_billing"]
|
|
|
|
|
|
def test_tools_to_session_tools_omits_disabled_function_tool():
|
|
tool = function_tool(
|
|
lambda: "ok",
|
|
name_override="hidden_tool",
|
|
is_enabled=False,
|
|
)
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
out = m._tools_to_session_tools([tool], [])
|
|
|
|
assert out == []
|
|
|
|
|
|
def test_tools_to_session_tools_ignores_disabled_handoff_name_conflict():
|
|
tool = function_tool(lambda: "ok", name_override="transfer_to_billing")
|
|
h = handoff(
|
|
Agent(name="billing"),
|
|
tool_name_override="transfer_to_billing",
|
|
is_enabled=False,
|
|
)
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
out = m._tools_to_session_tools([tool], [h])
|
|
|
|
assert [tool.name for tool in out] == ["transfer_to_billing"]
|
|
|
|
|
|
def test_tools_to_session_tools_rejects_duplicate_handoff_names():
|
|
handoff_one = handoff(Agent(name="billing"), tool_name_override="transfer_to_support")
|
|
handoff_two = handoff(Agent(name="technical"), tool_name_override="transfer_to_support")
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
with pytest.raises(
|
|
UserError,
|
|
match=("Duplicate Realtime tool name found: 'transfer_to_support' \\(2 handoffs\\)"),
|
|
):
|
|
m._tools_to_session_tools([], [handoff_one, handoff_two])
|
|
|
|
|
|
def test_tools_to_session_tools_rejects_namespaced_function_tools():
|
|
tool = tool_namespace(
|
|
name="crm",
|
|
description="CRM tools",
|
|
tools=[function_tool(lambda customer_id: customer_id, name_override="lookup_account")],
|
|
)[0]
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
with pytest.raises(UserError, match="tool_namespace\\(\\)"):
|
|
m._tools_to_session_tools([tool], [])
|
|
|
|
|
|
def test_tools_to_session_tools_rejects_deferred_function_tools():
|
|
tool = function_tool(
|
|
lambda customer_id: customer_id,
|
|
name_override="lookup_account",
|
|
defer_loading=True,
|
|
)
|
|
m = OpenAIRealtimeWebSocketModel()
|
|
|
|
with pytest.raises(UserError, match="defer_loading=True"):
|
|
m._tools_to_session_tools([tool], [])
|