test(llms): add unit tests for LLM adapters and utility modules (#154)
* test(llms): add unit tests for LLM adapters and utility modules Co-authored-by: Isaac * fix(test): address PR review — fix line length, docstrings, and CodeQL alert Co-authored-by: Isaac * fix: use explicit string concat, fix URL assertion Replace implicit adjacent-literal string concatenation with explicit `+` in test_anthropic_adapter.py to satisfy CodeQL. Replace URL substring checks with exact equality in test_vertex_adapter.py. Co-authored-by: Isaac
This commit is contained in:
Tomu Hirata
@@ -307,3 +307,251 @@ def test_string_user_content_passes_through() -> None:
|
||||
payload = _chat_to_anthropic(messages, "claude-test", None, {})
|
||||
# String content passed through as-is.
|
||||
assert payload["messages"][0]["content"] == "Hello"
|
||||
|
||||
|
||||
# ── Header building ──────────────────────────────────────
|
||||
|
||||
|
||||
def test_build_headers_with_api_key() -> None:
|
||||
"""API key is set in the x-api-key header."""
|
||||
from omnigent.llms.adapters.anthropic import _build_headers
|
||||
|
||||
headers = _build_headers(api_key_override="sk-test-123")
|
||||
assert headers["x-api-key"] == "sk-test-123"
|
||||
assert headers["anthropic-version"] == "2023-06-01"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
|
||||
|
||||
def test_build_headers_raises_without_api_key() -> None:
|
||||
"""Missing API key raises OmnigentError."""
|
||||
from omnigent.errors import OmnigentError
|
||||
from omnigent.llms.adapters.anthropic import _build_headers
|
||||
|
||||
with pytest.raises(OmnigentError, match="api_key"):
|
||||
_build_headers(api_key_override=None)
|
||||
|
||||
|
||||
def test_build_headers_raises_for_empty_api_key() -> None:
|
||||
"""Empty string API key raises OmnigentError."""
|
||||
from omnigent.errors import OmnigentError
|
||||
from omnigent.llms.adapters.anthropic import _build_headers
|
||||
|
||||
with pytest.raises(OmnigentError, match="api_key"):
|
||||
_build_headers(api_key_override="")
|
||||
|
||||
|
||||
# ── Reasoning effort ─────────────────────────────────────
|
||||
|
||||
|
||||
def test_effort_to_budget_low() -> None:
|
||||
from omnigent.llms.adapters.anthropic import _effort_to_budget
|
||||
|
||||
assert _effort_to_budget("low", 16384) == 1024
|
||||
|
||||
|
||||
def test_effort_to_budget_medium() -> None:
|
||||
from omnigent.llms.adapters.anthropic import _effort_to_budget
|
||||
|
||||
assert _effort_to_budget("medium", 16384) == 4096
|
||||
|
||||
|
||||
def test_effort_to_budget_high() -> None:
|
||||
from omnigent.llms.adapters.anthropic import _effort_to_budget
|
||||
|
||||
assert _effort_to_budget("high", 16384) == 8192
|
||||
|
||||
|
||||
def test_effort_to_budget_low_clamped_to_max_tokens() -> None:
|
||||
"""When max_tokens is less than the effort's budget, clamp to max_tokens."""
|
||||
from omnigent.llms.adapters.anthropic import _effort_to_budget
|
||||
|
||||
assert _effort_to_budget("low", 512) == 512
|
||||
|
||||
|
||||
def test_reasoning_effort_adds_thinking_to_payload() -> None:
|
||||
"""reasoning_effort in extra adds thinking config to the payload."""
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
payload = _chat_to_anthropic(messages, "claude-test", None, {"reasoning_effort": "high"})
|
||||
assert payload["thinking"]["type"] == "enabled"
|
||||
assert payload["thinking"]["budget_tokens"] == 8192
|
||||
|
||||
|
||||
# ── Stop sequences ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_stop_string_wrapped_in_list() -> None:
|
||||
"""A single stop string is wrapped in a list."""
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
payload = _chat_to_anthropic(messages, "claude-test", None, {"stop": "END"})
|
||||
assert payload["stop_sequences"] == ["END"]
|
||||
|
||||
|
||||
def test_stop_list_passed_through() -> None:
|
||||
"""A list of stop sequences passes through unchanged."""
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
payload = _chat_to_anthropic(messages, "claude-test", None, {"stop": ["END", "STOP"]})
|
||||
assert payload["stop_sequences"] == ["END", "STOP"]
|
||||
|
||||
|
||||
# ── Streaming SSE parsing ────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_to_chat_chunks_text_delta() -> None:
|
||||
"""Text deltas in the SSE stream produce Chat Completions chunks."""
|
||||
from omnigent.llms.adapters.anthropic import _stream_to_chat_chunks
|
||||
|
||||
lines = [
|
||||
"data: "
|
||||
+ '{"type": "message_start", "message": {"id": "msg_1",'
|
||||
+ ' "model": "claude-test",'
|
||||
+ ' "usage": {"input_tokens": 10}}}',
|
||||
'data: {"type": "content_block_start", "content_block": {"type": "text"}}',
|
||||
'data: {"type": "content_block_delta", "delta": {"type": "text_delta", "text": "Hello"}}',
|
||||
'data: {"type": "content_block_delta", "delta": {"type": "text_delta", "text": " world"}}',
|
||||
"data: "
|
||||
+ '{"type": "message_delta",'
|
||||
+ ' "delta": {"stop_reason": "end_turn"},'
|
||||
+ ' "usage": {"output_tokens": 5}}',
|
||||
]
|
||||
|
||||
async def _aiter():
|
||||
for line in lines:
|
||||
yield line
|
||||
|
||||
chunks = [c async for c in _stream_to_chat_chunks(_aiter())]
|
||||
# Two text delta chunks + one final chunk with usage
|
||||
text_chunks = [c for c in chunks if c["choices"][0]["delta"].get("content")]
|
||||
assert len(text_chunks) == 2
|
||||
assert text_chunks[0]["choices"][0]["delta"]["content"] == "Hello"
|
||||
assert text_chunks[1]["choices"][0]["delta"]["content"] == " world"
|
||||
|
||||
# Final chunk has usage
|
||||
final = chunks[-1]
|
||||
assert final["usage"]["prompt_tokens"] == 10
|
||||
assert final["usage"]["completion_tokens"] == 5
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_to_chat_chunks_tool_use() -> None:
|
||||
"""Tool use blocks in the SSE stream produce tool_calls in chunks."""
|
||||
from omnigent.llms.adapters.anthropic import _stream_to_chat_chunks
|
||||
|
||||
lines = [
|
||||
"data: "
|
||||
+ '{"type": "message_start", "message": {"id": "msg_2",'
|
||||
+ ' "model": "claude-test",'
|
||||
+ ' "usage": {"input_tokens": 5}}}',
|
||||
"data: "
|
||||
+ '{"type": "content_block_start",'
|
||||
+ ' "content_block": {"type": "tool_use",'
|
||||
+ ' "id": "tu_1", "name": "get_weather"}}',
|
||||
"data: "
|
||||
+ '{"type": "content_block_delta",'
|
||||
+ ' "delta": {"type": "input_json_delta",'
|
||||
+ ' "partial_json": "{\\"city\\":"}}',
|
||||
"data: "
|
||||
+ '{"type": "content_block_delta",'
|
||||
+ ' "delta": {"type": "input_json_delta",'
|
||||
+ ' "partial_json": "\\"London\\"}"}}',
|
||||
"data: "
|
||||
+ '{"type": "message_delta",'
|
||||
+ ' "delta": {"stop_reason": "tool_use"},'
|
||||
+ ' "usage": {"output_tokens": 10}}',
|
||||
]
|
||||
|
||||
async def _aiter():
|
||||
for line in lines:
|
||||
yield line
|
||||
|
||||
chunks = [c async for c in _stream_to_chat_chunks(_aiter())]
|
||||
# First chunk: tool_call start with id and name
|
||||
tool_start = chunks[0]
|
||||
tc = tool_start["choices"][0]["delta"]["tool_calls"][0]
|
||||
assert tc["id"] == "tu_1"
|
||||
assert tc["function"]["name"] == "get_weather"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_stream_skips_non_data_lines() -> None:
|
||||
"""Non-data lines are silently skipped."""
|
||||
from omnigent.llms.adapters.anthropic import _stream_to_chat_chunks
|
||||
|
||||
lines = [
|
||||
"event: message_start",
|
||||
"data: "
|
||||
+ '{"type": "message_start", "message":'
|
||||
+ ' {"id": "msg_3", "model": "claude-test",'
|
||||
+ ' "usage": {}}}',
|
||||
": comment line",
|
||||
"",
|
||||
"data: "
|
||||
+ '{"type": "message_delta",'
|
||||
+ ' "delta": {"stop_reason": "end_turn"},'
|
||||
+ ' "usage": {"output_tokens": 1}}',
|
||||
]
|
||||
|
||||
async def _aiter():
|
||||
for line in lines:
|
||||
yield line
|
||||
|
||||
chunks = [c async for c in _stream_to_chat_chunks(_aiter())]
|
||||
# Only the message_delta produces a chunk; message_start only sets metadata
|
||||
assert len(chunks) == 1
|
||||
|
||||
|
||||
# ── Tool choice edge case ────────────────────────────────
|
||||
|
||||
|
||||
def test_tool_choice_unknown_falls_back_to_auto() -> None:
|
||||
"""Unknown tool_choice values fall back to auto."""
|
||||
assert _convert_tool_choice("unknown_value") == {"type": "auto"}
|
||||
|
||||
|
||||
# ── Top P passthrough ────────────────────────────────────
|
||||
|
||||
|
||||
def test_top_p_passed_through() -> None:
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
payload = _chat_to_anthropic(messages, "claude-test", None, {"top_p": 0.9})
|
||||
assert payload["top_p"] == 0.9
|
||||
|
||||
|
||||
# ── Non-function tools skipped ───────────────────────────
|
||||
|
||||
|
||||
def test_non_function_tools_skipped() -> None:
|
||||
"""Non-function tool types are filtered out."""
|
||||
tools = [
|
||||
{"type": "not_function", "whatever": {}},
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "real_fn",
|
||||
"parameters": {},
|
||||
},
|
||||
},
|
||||
]
|
||||
result = _convert_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert result[0]["name"] == "real_fn"
|
||||
|
||||
|
||||
# ── Unrecognized content part passthrough ────────────────
|
||||
|
||||
|
||||
def test_unrecognized_part_passes_through() -> None:
|
||||
"""Unrecognized content part types pass through as-is."""
|
||||
part = {"type": "input_audio", "data": "base64data"}
|
||||
result = _translate_part_to_anthropic(part)
|
||||
assert result is part
|
||||
|
||||
|
||||
# ── Max completion tokens alias ──────────────────────────
|
||||
|
||||
|
||||
def test_max_completion_tokens_alias() -> None:
|
||||
"""max_completion_tokens is an alias for max_tokens."""
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
payload = _chat_to_anthropic(messages, "claude-test", None, {"max_completion_tokens": 2048})
|
||||
assert payload["max_tokens"] == 2048
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
"""Tests for llms.adapters.base — ABC enforcement."""
|
||||
|
||||
from collections.abc import AsyncIterator
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
|
||||
from omnigent.llms.adapters.base import BaseAdapter
|
||||
|
||||
|
||||
def test_cannot_instantiate_base_adapter() -> None:
|
||||
"""BaseAdapter is abstract and cannot be instantiated directly."""
|
||||
with pytest.raises(TypeError, match="abstract"):
|
||||
BaseAdapter() # type: ignore[abstract]
|
||||
|
||||
|
||||
def test_subclass_must_implement_chat_completions() -> None:
|
||||
"""A subclass that does not implement chat_completions cannot be instantiated."""
|
||||
|
||||
class IncompleteAdapter(BaseAdapter):
|
||||
pass
|
||||
|
||||
with pytest.raises(TypeError, match="abstract"):
|
||||
IncompleteAdapter() # type: ignore[abstract]
|
||||
|
||||
|
||||
def test_concrete_subclass_can_be_instantiated() -> None:
|
||||
"""A complete subclass that implements chat_completions can be instantiated."""
|
||||
|
||||
class ConcreteAdapter(BaseAdapter):
|
||||
async def chat_completions(
|
||||
self,
|
||||
messages: list[dict[str, Any]],
|
||||
model: str,
|
||||
tools: list[dict[str, Any]] | None,
|
||||
stream: bool,
|
||||
extra: dict[str, Any],
|
||||
*,
|
||||
connection_params: dict[str, str] | None = None,
|
||||
timeout: int | None = None,
|
||||
) -> dict[str, Any] | AsyncIterator[dict[str, Any]]:
|
||||
return {"choices": []}
|
||||
|
||||
adapter = ConcreteAdapter()
|
||||
assert isinstance(adapter, BaseAdapter)
|
||||
@@ -396,3 +396,155 @@ def test_string_user_content_becomes_text_block() -> None:
|
||||
messages = [{"role": "user", "content": "Hello"}]
|
||||
converse_msgs, _ = _messages_to_converse(messages)
|
||||
assert converse_msgs[0]["content"] == [{"text": "Hello"}]
|
||||
|
||||
|
||||
# ── Streaming chunk helpers ──────────────────────────────
|
||||
|
||||
|
||||
def test_stream_text_chunk_structure() -> None:
|
||||
"""_stream_text_chunk builds a valid Chat Completions text delta chunk."""
|
||||
from omnigent.llms.adapters.bedrock import _stream_text_chunk
|
||||
|
||||
chunk = _stream_text_chunk("bedrock-model", "Hello")
|
||||
assert chunk["model"] == "bedrock-model"
|
||||
assert chunk["object"] == "chat.completion.chunk"
|
||||
assert chunk["choices"][0]["delta"]["content"] == "Hello"
|
||||
assert chunk["choices"][0]["finish_reason"] is None
|
||||
|
||||
|
||||
def test_stream_stop_chunk_structure() -> None:
|
||||
"""_stream_stop_chunk builds a valid stop chunk."""
|
||||
from omnigent.llms.adapters.bedrock import _stream_stop_chunk
|
||||
|
||||
chunk = _stream_stop_chunk("bedrock-model", "stop")
|
||||
assert chunk["choices"][0]["finish_reason"] == "stop"
|
||||
assert chunk["choices"][0]["delta"] == {}
|
||||
|
||||
|
||||
def test_stream_stop_chunk_tool_calls() -> None:
|
||||
"""_stream_stop_chunk for tool_use produces tool_calls finish reason."""
|
||||
from omnigent.llms.adapters.bedrock import _stream_stop_chunk
|
||||
|
||||
chunk = _stream_stop_chunk("bedrock-model", "tool_calls")
|
||||
assert chunk["choices"][0]["finish_reason"] == "tool_calls"
|
||||
|
||||
|
||||
def test_stream_usage_chunk_structure() -> None:
|
||||
"""_stream_usage_chunk builds a valid usage chunk."""
|
||||
from omnigent.llms.adapters.bedrock import _stream_usage_chunk
|
||||
|
||||
usage = {"inputTokens": 10, "outputTokens": 5, "totalTokens": 15}
|
||||
chunk = _stream_usage_chunk("bedrock-model", usage)
|
||||
assert chunk["usage"]["prompt_tokens"] == 10
|
||||
assert chunk["usage"]["completion_tokens"] == 5
|
||||
assert chunk["usage"]["total_tokens"] == 15
|
||||
|
||||
|
||||
# ── None content becomes empty blocks ────────────────────
|
||||
|
||||
|
||||
def test_none_content_becomes_empty_blocks() -> None:
|
||||
"""None content yields empty content block list."""
|
||||
from omnigent.llms.adapters.bedrock import _content_to_converse_blocks
|
||||
|
||||
assert _content_to_converse_blocks(None) == []
|
||||
|
||||
|
||||
# ── Unrecognized part becomes text placeholder ───────────
|
||||
|
||||
|
||||
def test_unrecognized_part_becomes_text_placeholder() -> None:
|
||||
"""Unrecognized content part types render as text placeholder."""
|
||||
result = _translate_part_to_converse({"type": "input_audio", "data": "base64"})
|
||||
assert result == {"text": "[unsupported content: input_audio]"}
|
||||
|
||||
|
||||
# ── file_data without data URI raises ────────────────────
|
||||
|
||||
|
||||
def test_file_data_without_data_uri_raises() -> None:
|
||||
"""input_file without a data: URI prefix raises ValueError."""
|
||||
part = {
|
||||
"type": "input_file",
|
||||
"file_data": "https://example.com/file.pdf",
|
||||
}
|
||||
with pytest.raises(ValueError, match="data: URI"):
|
||||
_translate_part_to_converse(part)
|
||||
|
||||
|
||||
# ── file_data without filename ───────────────────────────
|
||||
|
||||
|
||||
def test_file_data_without_filename() -> None:
|
||||
"""input_file without filename omits name from document block."""
|
||||
part = {
|
||||
"type": "input_file",
|
||||
"file_data": "data:application/pdf;base64,JVBERi0xLjQK",
|
||||
}
|
||||
result = _translate_part_to_converse(part)
|
||||
assert "name" not in result["document"]
|
||||
|
||||
|
||||
# ── Non-function tools skipped ───────────────────────────
|
||||
|
||||
|
||||
def test_non_function_tools_skipped() -> None:
|
||||
"""Non-function tool types are filtered out."""
|
||||
tools = [
|
||||
{"type": "not_function", "whatever": {}},
|
||||
{"type": "function", "function": {"name": "fn", "parameters": {}}},
|
||||
]
|
||||
result = _convert_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert result[0]["toolSpec"]["name"] == "fn"
|
||||
|
||||
|
||||
# ── Tool without description ─────────────────────────────
|
||||
|
||||
|
||||
def test_tool_without_description_omits_description() -> None:
|
||||
"""Tool specs without description omit the field."""
|
||||
tools = [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {"name": "fn", "parameters": {}},
|
||||
}
|
||||
]
|
||||
result = _convert_tools(tools)
|
||||
assert "description" not in result[0]["toolSpec"]
|
||||
|
||||
|
||||
# ── System prompts None when absent ──────────────────────
|
||||
|
||||
|
||||
def test_no_system_messages_returns_none_system() -> None:
|
||||
"""No system messages -> system_prompts is None."""
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
_, system_prompts = _messages_to_converse(messages)
|
||||
assert system_prompts is None
|
||||
|
||||
|
||||
# ── Converse response with empty content ─────────────────
|
||||
|
||||
|
||||
def test_converse_response_no_content() -> None:
|
||||
"""Response with no text or tool content yields None content and empty tool_calls."""
|
||||
response = {
|
||||
"output": {"message": {"role": "assistant", "content": []}},
|
||||
"stopReason": "end_turn",
|
||||
"usage": {},
|
||||
}
|
||||
chat = _converse_to_chat(response, "bedrock-model")
|
||||
assert chat["choices"][0]["message"]["content"] is None
|
||||
assert chat["choices"][0]["message"]["tool_calls"] is None
|
||||
|
||||
|
||||
# ── max_completion_tokens alias ──────────────────────────
|
||||
|
||||
|
||||
def test_max_completion_tokens_alias() -> None:
|
||||
"""max_completion_tokens is an alias for max_tokens in inference config."""
|
||||
messages = [{"role": "user", "content": "Hi"}]
|
||||
extra = {"max_completion_tokens": 2048}
|
||||
kwargs = _build_converse_kwargs(messages, "model-id", None, extra)
|
||||
assert kwargs["inferenceConfig"]["maxTokens"] == 2048
|
||||
|
||||
@@ -305,3 +305,198 @@ def test_string_user_content_becomes_text_part() -> None:
|
||||
messages = [{"role": "user", "content": "Hello"}]
|
||||
payload = _chat_to_gemini(messages, None, {})
|
||||
assert payload["contents"][0]["parts"] == [{"text": "Hello"}]
|
||||
|
||||
|
||||
# ── Streaming chunk translation ──────────────────────────
|
||||
|
||||
|
||||
def test_gemini_stream_text_chunk() -> None:
|
||||
"""A streaming chunk with text produces a Chat Completions text delta."""
|
||||
from omnigent.llms.adapters.gemini import _gemini_stream_chunk_to_chat
|
||||
|
||||
data = {
|
||||
"candidates": [
|
||||
{
|
||||
"content": {"parts": [{"text": "Hello"}], "role": "model"},
|
||||
}
|
||||
],
|
||||
}
|
||||
chunks = list(_gemini_stream_chunk_to_chat(data))
|
||||
assert len(chunks) == 1
|
||||
assert chunks[0]["choices"][0]["delta"]["content"] == "Hello"
|
||||
assert chunks[0]["choices"][0]["finish_reason"] is None
|
||||
|
||||
|
||||
def test_gemini_stream_function_call_chunk() -> None:
|
||||
"""A streaming chunk with functionCall produces a tool_calls delta."""
|
||||
from omnigent.llms.adapters.gemini import _gemini_stream_chunk_to_chat
|
||||
|
||||
data = {
|
||||
"candidates": [
|
||||
{
|
||||
"content": {
|
||||
"parts": [
|
||||
{
|
||||
"functionCall": {
|
||||
"name": "get_weather",
|
||||
"args": {"city": "London"},
|
||||
}
|
||||
}
|
||||
],
|
||||
"role": "model",
|
||||
},
|
||||
}
|
||||
],
|
||||
}
|
||||
chunks = list(_gemini_stream_chunk_to_chat(data))
|
||||
assert len(chunks) == 1
|
||||
tc = chunks[0]["choices"][0]["delta"]["tool_calls"][0]
|
||||
assert tc["function"]["name"] == "get_weather"
|
||||
assert json.loads(tc["function"]["arguments"]) == {"city": "London"}
|
||||
|
||||
|
||||
def test_gemini_stream_finish_reason_chunk() -> None:
|
||||
"""A streaming chunk with finishReason emits a separate finish chunk."""
|
||||
from omnigent.llms.adapters.gemini import _gemini_stream_chunk_to_chat
|
||||
|
||||
data = {
|
||||
"candidates": [
|
||||
{
|
||||
"content": {"parts": [{"text": "Done"}], "role": "model"},
|
||||
"finishReason": "STOP",
|
||||
}
|
||||
],
|
||||
}
|
||||
chunks = list(_gemini_stream_chunk_to_chat(data))
|
||||
# Text chunk + finish reason chunk
|
||||
assert len(chunks) == 2
|
||||
assert chunks[1]["choices"][0]["finish_reason"] == "stop"
|
||||
|
||||
|
||||
def test_gemini_stream_usage_only_chunk() -> None:
|
||||
"""A streaming chunk with no candidates but usageMetadata yields usage."""
|
||||
from omnigent.llms.adapters.gemini import _gemini_stream_chunk_to_chat
|
||||
|
||||
data = {
|
||||
"usageMetadata": {
|
||||
"promptTokenCount": 10,
|
||||
"candidatesTokenCount": 5,
|
||||
"totalTokenCount": 15,
|
||||
},
|
||||
}
|
||||
chunks = list(_gemini_stream_chunk_to_chat(data))
|
||||
assert len(chunks) == 1
|
||||
assert chunks[0]["usage"]["prompt_tokens"] == 10
|
||||
assert chunks[0]["usage"]["completion_tokens"] == 5
|
||||
|
||||
|
||||
def test_gemini_stream_empty_candidates_no_usage() -> None:
|
||||
"""A streaming chunk with empty candidates and no usage yields nothing."""
|
||||
from omnigent.llms.adapters.gemini import _gemini_stream_chunk_to_chat
|
||||
|
||||
chunks = list(_gemini_stream_chunk_to_chat({"candidates": []}))
|
||||
assert chunks == []
|
||||
|
||||
|
||||
# ── Empty response ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_empty_chat_response_structure() -> None:
|
||||
"""_empty_chat_response returns a well-formed empty response."""
|
||||
from omnigent.llms.adapters.gemini import _empty_chat_response
|
||||
|
||||
resp = _empty_chat_response("gemini-test")
|
||||
assert resp["model"] == "gemini-test"
|
||||
assert resp["choices"][0]["message"]["content"] is None
|
||||
assert resp["choices"][0]["message"]["tool_calls"] is None
|
||||
assert resp["choices"][0]["finish_reason"] == "stop"
|
||||
|
||||
|
||||
# ── None content becomes empty parts ────────────────────
|
||||
|
||||
|
||||
def test_none_content_becomes_empty_parts() -> None:
|
||||
"""None content (e.g. assistant with tool_calls only) yields empty parts."""
|
||||
from omnigent.llms.adapters.gemini import _content_to_gemini_parts
|
||||
|
||||
assert _content_to_gemini_parts(None) == []
|
||||
|
||||
|
||||
# ── Gemini headers ───────────────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_headers_with_api_key() -> None:
|
||||
"""API key is set in x-goog-api-key header."""
|
||||
from omnigent.llms.adapters.gemini import GeminiAdapter
|
||||
|
||||
adapter = GeminiAdapter()
|
||||
headers = await adapter._get_headers(api_key_override="test-key")
|
||||
assert headers["x-goog-api-key"] == "test-key"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_headers_raises_without_api_key() -> None:
|
||||
"""Missing API key raises OmnigentError."""
|
||||
from omnigent.errors import OmnigentError
|
||||
from omnigent.llms.adapters.gemini import GeminiAdapter
|
||||
|
||||
adapter = GeminiAdapter()
|
||||
with pytest.raises(OmnigentError, match="api_key"):
|
||||
await adapter._get_headers(api_key_override=None)
|
||||
|
||||
|
||||
# ── Tool without description ─────────────────────────────
|
||||
|
||||
|
||||
def test_tool_without_description_omits_description() -> None:
|
||||
"""Tool declarations without description omit the field."""
|
||||
tools = [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "fn",
|
||||
"parameters": {},
|
||||
},
|
||||
}
|
||||
]
|
||||
result = _convert_tools(tools)
|
||||
assert "description" not in result[0]
|
||||
|
||||
|
||||
# ── Non-function tools skipped ───────────────────────────
|
||||
|
||||
|
||||
def test_non_function_tools_skipped() -> None:
|
||||
"""Non-function tool types are filtered out."""
|
||||
tools = [
|
||||
{"type": "not_function", "whatever": {}},
|
||||
{"type": "function", "function": {"name": "fn", "parameters": {}}},
|
||||
]
|
||||
result = _convert_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert result[0]["name"] == "fn"
|
||||
|
||||
|
||||
# ── Unrecognized part passthrough ────────────────────────
|
||||
|
||||
|
||||
def test_unrecognized_part_passes_through() -> None:
|
||||
"""Unrecognized content part types pass through as-is."""
|
||||
part = {"type": "input_audio", "data": "base64data"}
|
||||
result = _translate_part_to_gemini(part)
|
||||
assert result is part
|
||||
|
||||
|
||||
# ── file_data without data URI raises ────────────────────
|
||||
|
||||
|
||||
def test_file_data_without_data_uri_raises() -> None:
|
||||
"""input_file without a data: URI prefix raises ValueError."""
|
||||
part = {
|
||||
"type": "input_file",
|
||||
"file_data": "https://example.com/file.pdf",
|
||||
}
|
||||
with pytest.raises(ValueError, match="data: URI"):
|
||||
_translate_part_to_gemini(part)
|
||||
|
||||
@@ -5,6 +5,7 @@ import json
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from omnigent.llms.adapters.openai import (
|
||||
OpenAIAdapter,
|
||||
@@ -276,3 +277,312 @@ def test_stream_responses_decodes_utf8_split_across_chunks() -> None:
|
||||
return "".join(deltas)
|
||||
|
||||
assert asyncio.run(_run()) == "café"
|
||||
|
||||
|
||||
# ── URL resolution ──────────────────────────────────────
|
||||
|
||||
|
||||
def test_resolve_base_url_override_wins() -> None:
|
||||
from omnigent.llms.adapters.openai import _resolve_base_url
|
||||
|
||||
assert (
|
||||
_resolve_base_url("https://custom.api/v1/", "https://default.api/v1")
|
||||
== "https://custom.api/v1"
|
||||
)
|
||||
|
||||
|
||||
def test_resolve_base_url_falls_back_to_default() -> None:
|
||||
from omnigent.llms.adapters.openai import _resolve_base_url
|
||||
|
||||
assert _resolve_base_url(None, "https://default.api/v1") == "https://default.api/v1"
|
||||
|
||||
|
||||
def test_resolve_base_url_raises_when_both_none() -> None:
|
||||
from omnigent.errors import OmnigentError
|
||||
from omnigent.llms.adapters.openai import _resolve_base_url
|
||||
|
||||
with pytest.raises(OmnigentError, match="base_url"):
|
||||
_resolve_base_url(None, None)
|
||||
|
||||
|
||||
# ── Responses API tool conversion ───────────────────────
|
||||
|
||||
|
||||
def test_to_responses_tools_flattens_chat_format() -> None:
|
||||
from omnigent.llms.adapters.openai import _to_responses_tools
|
||||
|
||||
tools = [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
},
|
||||
}
|
||||
]
|
||||
result = _to_responses_tools(tools)
|
||||
assert len(result) == 1
|
||||
assert result[0] == {
|
||||
"type": "function",
|
||||
"name": "get_weather",
|
||||
"description": "Get the weather",
|
||||
"parameters": {"type": "object", "properties": {}},
|
||||
}
|
||||
|
||||
|
||||
def test_to_responses_tools_passes_through_responses_format() -> None:
|
||||
from omnigent.llms.adapters.openai import _to_responses_tools
|
||||
|
||||
tools = [
|
||||
{
|
||||
"type": "function",
|
||||
"name": "already_flat",
|
||||
"parameters": {},
|
||||
}
|
||||
]
|
||||
result = _to_responses_tools(tools)
|
||||
assert result[0] is tools[0]
|
||||
|
||||
|
||||
def test_to_responses_tools_no_description() -> None:
|
||||
from omnigent.llms.adapters.openai import _to_responses_tools
|
||||
|
||||
tools = [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "fn",
|
||||
"parameters": {},
|
||||
},
|
||||
}
|
||||
]
|
||||
result = _to_responses_tools(tools)
|
||||
assert "description" not in result[0]
|
||||
|
||||
|
||||
# ── Responses API output parsing ────────────────────────
|
||||
|
||||
|
||||
def test_parse_responses_output_message() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_output
|
||||
from omnigent.llms.types import MessageOutput
|
||||
|
||||
items = [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [{"type": "output_text", "text": "Hello!"}],
|
||||
}
|
||||
]
|
||||
output = _parse_responses_output(items)
|
||||
assert len(output) == 1
|
||||
assert isinstance(output[0], MessageOutput)
|
||||
assert output[0].content[0].text == "Hello!"
|
||||
|
||||
|
||||
def test_parse_responses_output_function_call() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_output
|
||||
from omnigent.llms.types import FunctionCallOutput
|
||||
|
||||
items = [
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "call_1",
|
||||
"name": "get_weather",
|
||||
"arguments": '{"city": "London"}',
|
||||
}
|
||||
]
|
||||
output = _parse_responses_output(items)
|
||||
assert len(output) == 1
|
||||
assert isinstance(output[0], FunctionCallOutput)
|
||||
assert output[0].call_id == "call_1"
|
||||
assert output[0].name == "get_weather"
|
||||
|
||||
|
||||
def test_parse_responses_output_native_tool() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_output
|
||||
from omnigent.llms.types import NativeToolOutput
|
||||
|
||||
items = [
|
||||
{
|
||||
"type": "web_search_call",
|
||||
"id": "ws_1",
|
||||
"status": "completed",
|
||||
}
|
||||
]
|
||||
output = _parse_responses_output(items)
|
||||
assert len(output) == 1
|
||||
assert isinstance(output[0], NativeToolOutput)
|
||||
assert output[0].data["type"] == "web_search_call"
|
||||
|
||||
|
||||
def test_parse_responses_output_reasoning_item() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_output
|
||||
from omnigent.llms.types import NativeToolOutput
|
||||
|
||||
items = [
|
||||
{
|
||||
"type": "reasoning",
|
||||
"content": [{"type": "text", "text": "thinking..."}],
|
||||
}
|
||||
]
|
||||
output = _parse_responses_output(items)
|
||||
assert len(output) == 1
|
||||
assert isinstance(output[0], NativeToolOutput)
|
||||
|
||||
|
||||
def test_parse_responses_output_ignores_unknown_type() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_output
|
||||
|
||||
items = [{"type": "unknown_future_type", "data": "something"}]
|
||||
output = _parse_responses_output(items)
|
||||
assert len(output) == 0
|
||||
|
||||
|
||||
# ── Responses API response parsing ──────────────────────
|
||||
|
||||
|
||||
def test_parse_responses_response_full() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_response
|
||||
from omnigent.llms.types import MessageOutput, Usage
|
||||
|
||||
data = {
|
||||
"model": "gpt-5.4",
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [{"type": "output_text", "text": "Hi"}],
|
||||
}
|
||||
],
|
||||
"usage": {
|
||||
"input_tokens": 10,
|
||||
"output_tokens": 5,
|
||||
"total_tokens": 15,
|
||||
},
|
||||
}
|
||||
resp = _parse_responses_response(data)
|
||||
assert resp.model == "gpt-5.4"
|
||||
assert isinstance(resp.output[0], MessageOutput)
|
||||
assert resp.usage == Usage(input_tokens=10, output_tokens=5, total_tokens=15)
|
||||
|
||||
|
||||
def test_parse_responses_response_missing_model_raises() -> None:
|
||||
from omnigent.errors import OmnigentError
|
||||
from omnigent.llms.adapters.openai import _parse_responses_response
|
||||
|
||||
with pytest.raises(OmnigentError, match="model"):
|
||||
_parse_responses_response({"output": []})
|
||||
|
||||
|
||||
def test_parse_responses_response_no_usage() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_response
|
||||
|
||||
data = {"model": "gpt-5.4", "output": []}
|
||||
resp = _parse_responses_response(data)
|
||||
assert resp.usage is None
|
||||
|
||||
|
||||
# ── Responses API SSE event parsing ─────────────────────
|
||||
|
||||
|
||||
def test_parse_responses_event_text_delta() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import ResponseTextDeltaEvent
|
||||
|
||||
event = _parse_responses_event("response.output_text.delta", {"delta": "Hello"})
|
||||
assert isinstance(event, ResponseTextDeltaEvent)
|
||||
assert event.delta == "Hello"
|
||||
|
||||
|
||||
def test_parse_responses_event_reasoning_delta() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import ResponseReasoningTextDeltaEvent
|
||||
|
||||
event = _parse_responses_event("response.reasoning_text.delta", {"delta": "thinking"})
|
||||
assert isinstance(event, ResponseReasoningTextDeltaEvent)
|
||||
assert event.delta == "thinking"
|
||||
|
||||
|
||||
def test_parse_responses_event_reasoning_summary_delta() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import ResponseReasoningSummaryTextDeltaEvent
|
||||
|
||||
event = _parse_responses_event("response.reasoning_summary_text.delta", {"delta": "summary"})
|
||||
assert isinstance(event, ResponseReasoningSummaryTextDeltaEvent)
|
||||
assert event.delta == "summary"
|
||||
|
||||
|
||||
def test_parse_responses_event_reasoning_started() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import ResponseReasoningStartedEvent
|
||||
|
||||
event = _parse_responses_event("response.output_item.added", {"item": {"type": "reasoning"}})
|
||||
assert isinstance(event, ResponseReasoningStartedEvent)
|
||||
|
||||
|
||||
def test_parse_responses_event_native_tool_done() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import NativeToolOutputAddedEvent
|
||||
|
||||
event = _parse_responses_event(
|
||||
"response.output_item.done",
|
||||
{"item": {"type": "web_search_call", "id": "ws_1", "status": "completed"}},
|
||||
)
|
||||
assert isinstance(event, NativeToolOutputAddedEvent)
|
||||
assert event.item["type"] == "web_search_call"
|
||||
|
||||
|
||||
def test_parse_responses_event_reasoning_done() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import NativeToolOutputAddedEvent
|
||||
|
||||
event = _parse_responses_event(
|
||||
"response.output_item.done",
|
||||
{"item": {"type": "reasoning", "content": []}},
|
||||
)
|
||||
assert isinstance(event, NativeToolOutputAddedEvent)
|
||||
|
||||
|
||||
def test_parse_responses_event_completed() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
from omnigent.llms.types import ResponseCompletedEvent
|
||||
|
||||
event = _parse_responses_event(
|
||||
"response.completed",
|
||||
{
|
||||
"response": {
|
||||
"model": "gpt-5.4",
|
||||
"output": [
|
||||
{
|
||||
"type": "message",
|
||||
"content": [{"type": "output_text", "text": "Done"}],
|
||||
}
|
||||
],
|
||||
"usage": {"input_tokens": 1, "output_tokens": 1, "total_tokens": 2},
|
||||
}
|
||||
},
|
||||
)
|
||||
assert isinstance(event, ResponseCompletedEvent)
|
||||
assert event.response.model == "gpt-5.4"
|
||||
|
||||
|
||||
def test_parse_responses_event_unknown_returns_none() -> None:
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
|
||||
assert _parse_responses_event("response.some_future_event", {}) is None
|
||||
|
||||
|
||||
def test_parse_responses_event_non_reasoning_item_added_returns_none() -> None:
|
||||
"""output_item.added for non-reasoning types returns None."""
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
|
||||
event = _parse_responses_event("response.output_item.added", {"item": {"type": "message"}})
|
||||
assert event is None
|
||||
|
||||
|
||||
def test_parse_responses_event_non_native_item_done_returns_none() -> None:
|
||||
"""output_item.done for non-native types returns None."""
|
||||
from omnigent.llms.adapters.openai import _parse_responses_event
|
||||
|
||||
event = _parse_responses_event("response.output_item.done", {"item": {"type": "message"}})
|
||||
assert event is None
|
||||
|
||||
@@ -672,3 +672,152 @@ async def test_kimi_reasoning_started_emitted_once_per_run() -> None:
|
||||
|
||||
assert len(reasoning_started) == 1
|
||||
assert [e.delta for e in reasoning_deltas] == ["part 1", " part 2"]
|
||||
|
||||
|
||||
# ── _extract_delta_content unit tests ──────────────────────────────
|
||||
|
||||
|
||||
def test_extract_delta_content_plain_string() -> None:
|
||||
"""Plain string content returns (text, empty_reasoning)."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
text, reasoning = _extract_delta_content("Hello")
|
||||
assert text == "Hello"
|
||||
assert reasoning == ""
|
||||
|
||||
|
||||
def test_extract_delta_content_non_string_non_list() -> None:
|
||||
"""Non-string, non-list content returns empty strings."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
text, reasoning = _extract_delta_content(42) # type: ignore[arg-type]
|
||||
assert text == ""
|
||||
assert reasoning == ""
|
||||
|
||||
|
||||
def test_extract_delta_content_list_with_text_blocks() -> None:
|
||||
"""List of text blocks extracts text."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
content = [{"type": "text", "text": "Hello"}, {"type": "text", "text": " world"}]
|
||||
text, reasoning = _extract_delta_content(content)
|
||||
assert text == "Hello world"
|
||||
assert reasoning == ""
|
||||
|
||||
|
||||
def test_extract_delta_content_list_with_output_text_blocks() -> None:
|
||||
"""output_text blocks also count as text."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
content = [{"type": "output_text", "text": "Hello"}]
|
||||
text, _reasoning = _extract_delta_content(content)
|
||||
assert text == "Hello"
|
||||
|
||||
|
||||
def test_extract_delta_content_list_with_reasoning_blocks() -> None:
|
||||
"""Reasoning blocks extract summary text into reasoning output."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
content = [
|
||||
{
|
||||
"type": "reasoning",
|
||||
"summary": [{"type": "summary_text", "text": "thinking..."}],
|
||||
}
|
||||
]
|
||||
text, reasoning = _extract_delta_content(content)
|
||||
assert text == ""
|
||||
assert reasoning == "thinking..."
|
||||
|
||||
|
||||
def test_extract_delta_content_list_with_bare_strings() -> None:
|
||||
"""Bare strings in the list are treated as text."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
content = ["Hello", " world"]
|
||||
text, _reasoning = _extract_delta_content(content)
|
||||
assert text == "Hello world"
|
||||
|
||||
|
||||
def test_extract_delta_content_list_skips_non_dict_non_string() -> None:
|
||||
"""Non-dict, non-string items in the list are skipped."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
content = [42, {"type": "text", "text": "ok"}]
|
||||
text, _reasoning = _extract_delta_content(content)
|
||||
assert text == "ok"
|
||||
|
||||
|
||||
def test_extract_delta_content_reasoning_without_summary() -> None:
|
||||
"""Reasoning block without summary key yields no reasoning text."""
|
||||
from omnigent.llms._responses_to_chat import _extract_delta_content
|
||||
|
||||
content = [{"type": "reasoning"}]
|
||||
_text, reasoning = _extract_delta_content(content)
|
||||
assert reasoning == ""
|
||||
|
||||
|
||||
# ── _extract_usage unit tests ──────────────────────────────────────
|
||||
|
||||
|
||||
def test_extract_usage_returns_none_for_none() -> None:
|
||||
from omnigent.llms._responses_to_chat import _extract_usage
|
||||
|
||||
assert _extract_usage(None) is None
|
||||
|
||||
|
||||
def test_extract_usage_returns_none_for_empty_dict() -> None:
|
||||
from omnigent.llms._responses_to_chat import _extract_usage
|
||||
|
||||
assert _extract_usage({}) is None
|
||||
|
||||
|
||||
def test_extract_usage_maps_fields() -> None:
|
||||
from omnigent.llms._responses_to_chat import _extract_usage
|
||||
|
||||
usage = _extract_usage({"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15})
|
||||
assert usage is not None
|
||||
assert usage.input_tokens == 10
|
||||
assert usage.output_tokens == 5
|
||||
assert usage.total_tokens == 15
|
||||
|
||||
|
||||
# ── Streaming: usage-only final chunk ──────────────────────────────
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_usage_only_chunk() -> None:
|
||||
"""A trailing chunk with only usage (no choices) captures the usage."""
|
||||
chunks = [
|
||||
{"choices": [{"delta": {"content": "Hi"}, "finish_reason": None}]},
|
||||
{"choices": [{"delta": {}, "finish_reason": "stop"}]},
|
||||
{
|
||||
"usage": {
|
||||
"prompt_tokens": 10,
|
||||
"completion_tokens": 2,
|
||||
"total_tokens": 12,
|
||||
}
|
||||
},
|
||||
]
|
||||
events = [e async for e in chat_stream_to_response_events(_aiter(chunks), model="test")]
|
||||
completed = events[-1]
|
||||
assert isinstance(completed, ResponseCompletedEvent)
|
||||
assert completed.response.usage == Usage(input_tokens=10, output_tokens=2, total_tokens=12)
|
||||
|
||||
|
||||
# ── Trailing tool calls flushed ────────────────────────────────────
|
||||
|
||||
|
||||
def test_trailing_function_calls_flushed() -> None:
|
||||
"""Function call items at the end of input are flushed into assistant msg."""
|
||||
items = [
|
||||
{
|
||||
"type": "function_call",
|
||||
"call_id": "c1",
|
||||
"name": "fn",
|
||||
"arguments": "{}",
|
||||
},
|
||||
]
|
||||
messages = responses_input_to_chat_messages(items, None)
|
||||
assert len(messages) == 1
|
||||
assert messages[0]["role"] == "assistant"
|
||||
assert len(messages[0]["tool_calls"]) == 1
|
||||
|
||||
@@ -0,0 +1,175 @@
|
||||
"""Tests for llms.types — dataclass construction and edge cases."""
|
||||
|
||||
from omnigent.llms.types import (
|
||||
NATIVE_TOOL_OUTPUT_TYPES,
|
||||
FunctionCallOutput,
|
||||
MessageOutput,
|
||||
NativeToolOutput,
|
||||
NativeToolOutputAddedEvent,
|
||||
OutputText,
|
||||
Response,
|
||||
ResponseCompletedEvent,
|
||||
ResponseReasoningStartedEvent,
|
||||
ResponseReasoningSummaryTextDeltaEvent,
|
||||
ResponseReasoningTextDeltaEvent,
|
||||
ResponseTextDeltaEvent,
|
||||
Usage,
|
||||
)
|
||||
|
||||
# ── OutputText ──────────────────────────────────────────
|
||||
|
||||
|
||||
def test_output_text_defaults() -> None:
|
||||
ot = OutputText(text="Hello")
|
||||
assert ot.text == "Hello"
|
||||
assert ot.type == "output_text"
|
||||
assert ot.annotations is None
|
||||
|
||||
|
||||
def test_output_text_with_annotations() -> None:
|
||||
annotations = [{"type": "file_citation", "file_id": "f1"}]
|
||||
ot = OutputText(text="Hello", annotations=annotations)
|
||||
assert ot.annotations == annotations
|
||||
|
||||
|
||||
# ── MessageOutput ───────────────────────────────────────
|
||||
|
||||
|
||||
def test_message_output_defaults() -> None:
|
||||
mo = MessageOutput(content=[OutputText(text="Hi")])
|
||||
assert mo.type == "message"
|
||||
assert len(mo.content) == 1
|
||||
assert mo.content[0].text == "Hi"
|
||||
|
||||
|
||||
# ── FunctionCallOutput ──────────────────────────────────
|
||||
|
||||
|
||||
def test_function_call_output_defaults() -> None:
|
||||
fc = FunctionCallOutput(call_id="c1", name="fn", arguments="{}")
|
||||
assert fc.type == "function_call"
|
||||
assert fc.call_id == "c1"
|
||||
assert fc.name == "fn"
|
||||
assert fc.arguments == "{}"
|
||||
|
||||
|
||||
# ── NativeToolOutput ────────────────────────────────────
|
||||
|
||||
|
||||
def test_native_tool_output() -> None:
|
||||
data = {"type": "web_search_call", "id": "ws_1", "status": "completed"}
|
||||
nto = NativeToolOutput(data=data)
|
||||
assert nto.data["type"] == "web_search_call"
|
||||
|
||||
|
||||
# ── NATIVE_TOOL_OUTPUT_TYPES ────────────────────────────
|
||||
|
||||
|
||||
def test_native_tool_output_types_is_frozenset() -> None:
|
||||
assert isinstance(NATIVE_TOOL_OUTPUT_TYPES, frozenset)
|
||||
assert "web_search_call" in NATIVE_TOOL_OUTPUT_TYPES
|
||||
assert "file_search_call" in NATIVE_TOOL_OUTPUT_TYPES
|
||||
assert "code_interpreter_call" in NATIVE_TOOL_OUTPUT_TYPES
|
||||
assert "mcp_call" in NATIVE_TOOL_OUTPUT_TYPES
|
||||
|
||||
|
||||
# ── Usage ───────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_usage_defaults_to_none() -> None:
|
||||
u = Usage()
|
||||
assert u.input_tokens is None
|
||||
assert u.output_tokens is None
|
||||
assert u.total_tokens is None
|
||||
|
||||
|
||||
def test_usage_with_values() -> None:
|
||||
u = Usage(input_tokens=10, output_tokens=5, total_tokens=15)
|
||||
assert u.input_tokens == 10
|
||||
assert u.output_tokens == 5
|
||||
assert u.total_tokens == 15
|
||||
|
||||
|
||||
def test_usage_equality() -> None:
|
||||
a = Usage(input_tokens=10, output_tokens=5, total_tokens=15)
|
||||
b = Usage(input_tokens=10, output_tokens=5, total_tokens=15)
|
||||
assert a == b
|
||||
|
||||
|
||||
def test_usage_inequality() -> None:
|
||||
a = Usage(input_tokens=10, output_tokens=5, total_tokens=15)
|
||||
b = Usage(input_tokens=10, output_tokens=6, total_tokens=16)
|
||||
assert a != b
|
||||
|
||||
|
||||
# ── Response ────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_response_with_text() -> None:
|
||||
resp = Response(
|
||||
output=[MessageOutput(content=[OutputText(text="Hello")])],
|
||||
model="gpt-5.4",
|
||||
usage=Usage(input_tokens=10, output_tokens=5, total_tokens=15),
|
||||
)
|
||||
assert resp.model == "gpt-5.4"
|
||||
assert len(resp.output) == 1
|
||||
assert resp.usage is not None
|
||||
assert resp.usage.total_tokens == 15
|
||||
|
||||
|
||||
def test_response_no_usage() -> None:
|
||||
resp = Response(output=[], model="test-model")
|
||||
assert resp.usage is None
|
||||
|
||||
|
||||
def test_response_mixed_output() -> None:
|
||||
resp = Response(
|
||||
output=[
|
||||
MessageOutput(content=[OutputText(text="Hi")]),
|
||||
FunctionCallOutput(call_id="c1", name="fn", arguments="{}"),
|
||||
NativeToolOutput(data={"type": "web_search_call"}),
|
||||
],
|
||||
model="test-model",
|
||||
)
|
||||
assert len(resp.output) == 3
|
||||
assert isinstance(resp.output[0], MessageOutput)
|
||||
assert isinstance(resp.output[1], FunctionCallOutput)
|
||||
assert isinstance(resp.output[2], NativeToolOutput)
|
||||
|
||||
|
||||
# ── Streaming event types ───────────────────────────────
|
||||
|
||||
|
||||
def test_response_text_delta_event_defaults() -> None:
|
||||
e = ResponseTextDeltaEvent(delta="Hello")
|
||||
assert e.type == "response.output_text.delta"
|
||||
assert e.delta == "Hello"
|
||||
|
||||
|
||||
def test_response_reasoning_text_delta_event_defaults() -> None:
|
||||
e = ResponseReasoningTextDeltaEvent(delta="thinking")
|
||||
assert e.type == "response.reasoning_text.delta"
|
||||
|
||||
|
||||
def test_response_reasoning_summary_text_delta_event_defaults() -> None:
|
||||
e = ResponseReasoningSummaryTextDeltaEvent(delta="summary")
|
||||
assert e.type == "response.reasoning_summary_text.delta"
|
||||
|
||||
|
||||
def test_response_reasoning_started_event_defaults() -> None:
|
||||
e = ResponseReasoningStartedEvent()
|
||||
assert e.type == "response.reasoning.started"
|
||||
|
||||
|
||||
def test_native_tool_output_added_event_defaults() -> None:
|
||||
item = {"type": "web_search_call", "id": "ws_1"}
|
||||
e = NativeToolOutputAddedEvent(item=item)
|
||||
assert e.type == "response.output_item.done"
|
||||
assert e.item is item
|
||||
|
||||
|
||||
def test_response_completed_event_defaults() -> None:
|
||||
resp = Response(output=[], model="test")
|
||||
e = ResponseCompletedEvent(response=resp)
|
||||
assert e.type == "response.completed"
|
||||
assert e.response is resp
|
||||
@@ -447,3 +447,63 @@ def test_concurrent_notifies_do_not_lose_updates(
|
||||
assert len(written) == 1
|
||||
payload = json.loads(written[0].read_text())
|
||||
assert payload["totals"]["calls"] == threads * iters
|
||||
|
||||
|
||||
# ── notify_from_dict ─────────────────────────────────────────────
|
||||
|
||||
|
||||
def test_notify_from_dict_unpacks_usage(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
|
||||
"""notify_from_dict unpacks standard keys and delegates to notify."""
|
||||
monkeypatch.setenv(_usage_observer._ENV_VAR, str(tmp_path / "tokens.json"))
|
||||
_usage_observer.set_current_test("test_from_dict")
|
||||
_usage_observer.notify_from_dict(
|
||||
model="m",
|
||||
usage={"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
|
||||
)
|
||||
bucket = _usage_observer._RECORDS["test_from_dict"]
|
||||
assert bucket["total_tokens"] == 15
|
||||
assert bucket["calls"] == 1
|
||||
|
||||
|
||||
def test_notify_from_dict_none_usage_is_noop(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""notify_from_dict with None usage is a no-op."""
|
||||
monkeypatch.setenv(_usage_observer._ENV_VAR, str(tmp_path / "tokens.json"))
|
||||
_usage_observer.set_current_test("test_none")
|
||||
_usage_observer.notify_from_dict(model="m", usage=None)
|
||||
assert _usage_observer._RECORDS == {}
|
||||
|
||||
|
||||
def test_notify_from_dict_empty_dict_is_noop(
|
||||
monkeypatch: pytest.MonkeyPatch, tmp_path: Path
|
||||
) -> None:
|
||||
"""notify_from_dict({}) resolves to notify(..., 0, 0, 0) — records stay empty."""
|
||||
monkeypatch.setenv(_usage_observer._ENV_VAR, str(tmp_path / "tokens.json"))
|
||||
_usage_observer.set_current_test("test_empty")
|
||||
_usage_observer.notify_from_dict(model="m", usage={})
|
||||
assert _usage_observer._RECORDS == {}
|
||||
|
||||
|
||||
def test_notify_from_dict_non_dict_is_noop(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
"""notify_from_dict with a non-dict value is a no-op."""
|
||||
monkeypatch.delenv(_usage_observer._ENV_VAR, raising=False)
|
||||
_usage_observer.notify_from_dict(model="m", usage="not a dict") # type: ignore[arg-type]
|
||||
assert _usage_observer._RECORDS == {}
|
||||
|
||||
|
||||
# ── Double remove is idempotent ──────────────────────────────────
|
||||
|
||||
|
||||
def test_observer_remove_is_idempotent(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Calling remove() twice does not raise."""
|
||||
monkeypatch.delenv(_usage_observer._ENV_VAR, raising=False)
|
||||
|
||||
def cb(**_: Any) -> None:
|
||||
pass
|
||||
|
||||
remove = _usage_observer.add_observer(cb)
|
||||
remove()
|
||||
remove() # second call should not raise
|
||||
|
||||
@@ -86,3 +86,41 @@ def test_build_vertex_url_structure() -> None:
|
||||
"/locations/us-central1"
|
||||
"/publishers/google/models"
|
||||
)
|
||||
|
||||
|
||||
# ── _get_base_url raises ────────────────────────────────
|
||||
|
||||
|
||||
def test_get_base_url_raises() -> None:
|
||||
"""VertexAdapter._get_base_url always raises — Vertex requires connection_params."""
|
||||
from omnigent.llms.adapters.vertex import VertexAdapter
|
||||
|
||||
adapter = VertexAdapter()
|
||||
with pytest.raises(OmnigentError, match="requires"):
|
||||
adapter._get_base_url()
|
||||
|
||||
|
||||
# ── URL for different regions ────────────────────────────
|
||||
|
||||
|
||||
def test_build_vertex_url_different_region() -> None:
|
||||
"""URL changes with region."""
|
||||
url = _build_vertex_url("proj-2", "europe-west4")
|
||||
expected_url = (
|
||||
"https://europe-west4-aiplatform.googleapis.com"
|
||||
"/v1/projects/proj-2"
|
||||
"/locations/europe-west4"
|
||||
"/publishers/google/models"
|
||||
)
|
||||
assert url == expected_url
|
||||
|
||||
|
||||
# ── Resolve preserves extra keys ─────────────────────────
|
||||
|
||||
|
||||
def test_resolve_preserves_extra_connection_keys() -> None:
|
||||
"""Extra keys in connection_params are preserved after resolution."""
|
||||
params = {"project": "p", "location": "l", "extra_key": "value"}
|
||||
result = _resolve_vertex_params(params)
|
||||
assert result["extra_key"] == "value"
|
||||
assert "base_url" in result
|
||||
|
||||
Reference in New Issue
Block a user