"""Tests for src/openai_schemas.py Phase 2 of any_type_componentization_20260621. Verifies: - ToolCall + ToolCallFunction round-trip via to_dict - ChatMessage round-trip for all 4 roles - UsageStats field access - NormalizedResponse legacy dict preservation - OpenAICompatibleRequest typed messages - raw_response remains Any (Pattern 3 preserved) - tools field stays list[dict[str, Any]] for cross-phase Phase 1 ToolSpec (deferred to follow-up track per spec 3.4) CONVENTION: 1-space indentation. NO COMMENTS. """ from __future__ import annotations import json import pytest from src import openai_schemas def test_tool_call_function_construction() -> None: tcf = openai_schemas.ToolCallFunction(name="get_weather", arguments='{"city": "sf"}') assert tcf.name == "get_weather" assert tcf.arguments == '{"city": "sf"}' def test_tool_call_to_dict_round_trip() -> None: tc = openai_schemas.ToolCall( id="call_123", type="function", function=openai_schemas.ToolCallFunction(name="read_file", arguments='{"path": "/x.py"}'), ) d = tc.to_dict() assert d["id"] == "call_123" assert d["type"] == "function" assert d["function"]["name"] == "read_file" assert d["function"]["arguments"] == '{"path": "/x.py"}' def test_tool_call_defaults() -> None: tc = openai_schemas.ToolCall( id="call_x", function=openai_schemas.ToolCallFunction(name="noop", arguments="{}"), ) assert tc.type == "function" def test_tool_call_is_frozen() -> None: tc = openai_schemas.ToolCall( id="call_y", function=openai_schemas.ToolCallFunction(name="noop", arguments="{}"), ) with pytest.raises(Exception): tc.id = "mutated" def test_chat_message_system_role() -> None: msg = openai_schemas.ChatMessage(role="system", content="You are a helper.") d = msg.to_dict() assert d["role"] == "system" assert d["content"] == "You are a helper." assert "tool_calls" not in d assert "tool_call_id" not in d def test_chat_message_user_role() -> None: msg = openai_schemas.ChatMessage(role="user", content="Hello") d = msg.to_dict() assert d["role"] == "user" assert d["content"] == "Hello" def test_chat_message_assistant_with_tool_calls() -> None: tc = openai_schemas.ToolCall( id="call_a", function=openai_schemas.ToolCallFunction(name="read_file", arguments='{"path": "/x"}'), ) msg = openai_schemas.ChatMessage(role="assistant", content="", tool_calls=(tc,)) d = msg.to_dict() assert d["role"] == "assistant" assert d["content"] == "" assert len(d["tool_calls"]) == 1 assert d["tool_calls"][0]["function"]["name"] == "read_file" def test_chat_message_tool_role() -> None: msg = openai_schemas.ChatMessage( role="tool", content='{"result": "ok"}', tool_call_id="call_a" ) d = msg.to_dict() assert d["role"] == "tool" assert d["tool_call_id"] == "call_a" def test_chat_message_is_frozen() -> None: msg = openai_schemas.ChatMessage(role="user", content="hi") with pytest.raises(Exception): msg.role = "mutated" def test_usage_stats_construction() -> None: u = openai_schemas.UsageStats(input_tokens=100, output_tokens=50) assert u.input_tokens == 100 assert u.output_tokens == 50 assert u.cache_read_tokens == 0 assert u.cache_creation_tokens == 0 def test_usage_stats_with_cache() -> None: u = openai_schemas.UsageStats( input_tokens=100, output_tokens=50, cache_read_tokens=80, cache_creation_tokens=20, ) assert u.cache_read_tokens == 80 assert u.cache_creation_tokens == 20 def test_usage_stats_is_frozen() -> None: u = openai_schemas.UsageStats(input_tokens=1, output_tokens=1) with pytest.raises(Exception): u.input_tokens = 999 def test_normalized_response_construction() -> None: tc = openai_schemas.ToolCall( id="call_z", function=openai_schemas.ToolCallFunction(name="noop", arguments="{}"), ) usage = openai_schemas.UsageStats(input_tokens=10, output_tokens=20) resp = openai_schemas.NormalizedResponse( text="hello", tool_calls=(tc,), usage=usage, raw_response=None ) assert resp.text == "hello" assert len(resp.tool_calls) == 1 assert resp.usage.input_tokens == 10 assert resp.raw_response is None def test_normalized_response_raw_can_be_any_type() -> None: """Pattern 3: raw_response is intentionally Any (SDK-specific).""" usage = openai_schemas.UsageStats(input_tokens=0, output_tokens=0) resp = openai_schemas.NormalizedResponse( text="", tool_calls=(), usage=usage, raw_response={"vendor_specific": True} ) assert resp.raw_response == {"vendor_specific": True} def test_normalized_response_to_legacy_dict_preserves_shape() -> None: tc = openai_schemas.ToolCall( id="call_q", function=openai_schemas.ToolCallFunction(name="x", arguments="{}"), ) usage = openai_schemas.UsageStats( input_tokens=10, output_tokens=20, cache_read_tokens=5, cache_creation_tokens=3 ) resp = openai_schemas.NormalizedResponse( text="hello", tool_calls=(tc,), usage=usage, raw_response="sdk_obj" ) d = resp.to_legacy_dict() assert d["text"] == "hello" assert d["tool_calls"][0]["id"] == "call_q" assert d["usage"]["input_tokens"] == 10 assert d["usage"]["cache_read_tokens"] == 5 assert d["raw_response"] == "sdk_obj" def test_openai_compatible_request_defaults() -> None: msg = openai_schemas.ChatMessage(role="user", content="hi") req = openai_schemas.OpenAICompatibleRequest(messages=[msg], model="gpt-4") assert req.messages == [msg] assert req.model == "gpt-4" assert req.temperature == 0.0 assert req.top_p == 1.0 assert req.max_tokens == 8192 assert req.tools is None assert req.tool_choice == "auto" assert req.stream is False assert req.stream_callback is None assert req.extra_body is None def test_openai_compatible_request_tools_field_stays_dict_list() -> None: """Cross-phase coupling (deferred): Phase 1 ToolSpec migration is a follow-up track per spec 3.4. The tools field stays list[dict[str, Any]] for now.""" msg = openai_schemas.ChatMessage(role="user", content="hi") tools = [{"type": "function", "function": {"name": "x"}}] req = openai_schemas.OpenAICompatibleRequest(messages=[msg], model="gpt-4", tools=tools) assert req.tools == tools def test_chat_message_to_dict_handles_optional_fields() -> None: msg = openai_schemas.ChatMessage(role="assistant", content="", name=None, tool_call_id=None) d = msg.to_dict() assert "name" not in d assert "tool_call_id" not in d def test_normalized_response_is_frozen() -> None: usage = openai_schemas.UsageStats(input_tokens=0, output_tokens=0) resp = openai_schemas.NormalizedResponse(text="x", tool_calls=(), usage=usage, raw_response=None) with pytest.raises(Exception): resp.text = "mutated"