mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-11 08:38:24 +00:00
test: add handler helpers and tools extended tests
Add tests for: - handler.py static/helpers (resolve_max_tokens, build_followup, etc.) - tools.py uncovered (openai format, native detection, conversion) Extends test coverage for WebFetch interception feature.
This commit is contained in:
Thomas Menard
parent
2ad2c85cc3
commit
5fa3bdb4de
2 changed files with 433 additions and 0 deletions
307
tests/webfetch_tests/test_handler_helpers.py
Normal file
307
tests/webfetch_tests/test_handler_helpers.py
Normal file
|
|
@ -0,0 +1,307 @@
|
|||
"""Tests for handler.py uncovered methods.
|
||||
|
||||
Focus on static methods and internal helpers that can be tested
|
||||
without mocking the entire litellm ecosystem.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, AsyncMock, patch
|
||||
|
||||
from litellm.integrations.webfetch_interception.handler import WebFetchInterceptionLogger
|
||||
|
||||
|
||||
class TestResolveMaxTokens:
|
||||
"""Test _resolve_max_tokens helper."""
|
||||
|
||||
def test_default_value(self):
|
||||
"""Default max_tokens is 1024."""
|
||||
result = WebFetchInterceptionLogger._resolve_max_tokens({}, {})
|
||||
assert result == 1024
|
||||
|
||||
def test_from_optional_params(self):
|
||||
"""max_tokens from optional_params."""
|
||||
result = WebFetchInterceptionLogger._resolve_max_tokens(
|
||||
{"max_tokens": 2048}, {}
|
||||
)
|
||||
assert result == 2048
|
||||
|
||||
def test_from_kwargs(self):
|
||||
"""max_tokens from kwargs."""
|
||||
result = WebFetchInterceptionLogger._resolve_max_tokens(
|
||||
{}, {"max_tokens": 4096}
|
||||
)
|
||||
assert result == 4096
|
||||
|
||||
def test_thinking_budget_adjustment(self):
|
||||
"""max_tokens <= thinking.budget_tokens triggers adjustment."""
|
||||
result = WebFetchInterceptionLogger._resolve_max_tokens(
|
||||
{"max_tokens": 1000},
|
||||
{"max_tokens": 1000, "thinking": {"budget_tokens": 1500}},
|
||||
)
|
||||
# Should adjust to budget_tokens + 1024
|
||||
assert result == 2524
|
||||
|
||||
def test_thinking_budget_no_adjustment(self):
|
||||
"""max_tokens > thinking.budget_tokens no adjustment."""
|
||||
result = WebFetchInterceptionLogger._resolve_max_tokens(
|
||||
{"max_tokens": 5000},
|
||||
{"thinking": {"budget_tokens": 1000}},
|
||||
)
|
||||
assert result == 5000
|
||||
|
||||
|
||||
class TestPrepareFollowupKwargs:
|
||||
"""Test _prepare_followup_kwargs."""
|
||||
|
||||
def test_basic(self):
|
||||
"""Basic followup kwargs preparation."""
|
||||
kwargs = {"model": "claude-3", "messages": [], "max_tokens": 1000}
|
||||
result = WebFetchInterceptionLogger._prepare_followup_kwargs(kwargs)
|
||||
|
||||
assert "model" in result
|
||||
assert "api_key" in result
|
||||
assert "base_url" in result
|
||||
|
||||
def test_direct_params(self):
|
||||
"""Direct params overrides."""
|
||||
kwargs = {
|
||||
"model": "gpt-4",
|
||||
"api_key": "sk-test",
|
||||
"base_url": "https://custom.example.com",
|
||||
}
|
||||
result = WebFetchInterceptionLogger._prepare_followup_kwargs(kwargs)
|
||||
|
||||
assert result["api_key"] == "sk-test"
|
||||
assert result["base_url"] == "https://custom.example.com"
|
||||
|
||||
|
||||
class TestBuildFollowUpMessages:
|
||||
"""Test _build_follow_up_messages."""
|
||||
|
||||
def test_basic(self):
|
||||
"""Basic follow-up message building."""
|
||||
result = WebFetchInterceptionLogger._build_follow_up_messages(
|
||||
tool_call_id="call-1",
|
||||
function_name="web_fetch",
|
||||
fetch_result="Fetched content",
|
||||
)
|
||||
|
||||
assert result["tool_call_id"] == "call-1"
|
||||
assert result["role"] == "tool"
|
||||
assert "Fetched content" in result["content"]
|
||||
|
||||
def test_removes_thinking_blocks(self):
|
||||
"""Thinking content is stripped."""
|
||||
result = WebFetchInterceptionLogger._build_follow_up_messages(
|
||||
tool_call_id="call-1",
|
||||
function_name="fetch",
|
||||
fetch_result="<antThinking>thinking</antThinking>\nActual content",
|
||||
)
|
||||
|
||||
assert "<antThinking>" not in result["content"]
|
||||
assert "Actual content" in result["content"]
|
||||
|
||||
|
||||
class TestBuildKwargsForFollowup:
|
||||
"""Test _build_kwargs_for_followup."""
|
||||
|
||||
def test_basic(self):
|
||||
"""Build kwargs for followup request."""
|
||||
kwargs = {
|
||||
"model": "claude-3-opus-20240229",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"max_tokens": 4096,
|
||||
"optional_params": {"temperature": 0.7},
|
||||
}
|
||||
|
||||
result = WebFetchInterceptionLogger._build_kwargs_for_followup(kwargs)
|
||||
|
||||
assert result["model"] == "claude-3-opus-20240229"
|
||||
assert result["max_tokens"] == 4096
|
||||
|
||||
def test_resolves_model_name(self):
|
||||
"""Model name is resolved from router."""
|
||||
with patch(
|
||||
"litellm.integrations.webfetch_interception.handler.WebFetchInterceptionLogger._resolve_full_model_name",
|
||||
return_value="anthropic/claude-3-opus",
|
||||
):
|
||||
kwargs = {
|
||||
"model": "claude-opus",
|
||||
"messages": [],
|
||||
}
|
||||
result = WebFetchInterceptionLogger._build_kwargs_for_followup(kwargs)
|
||||
assert result["model"] == "anthropic/claude-3-opus"
|
||||
|
||||
|
||||
class TestResolveFullModelName:
|
||||
"""Test _resolve_full_model_name."""
|
||||
|
||||
def test_anthropic_prefix(self):
|
||||
"""Anthropic models get full prefix."""
|
||||
result = WebFetchInterceptionLogger._resolve_full_model_name(
|
||||
"claude-3-opus",
|
||||
{"custom_llm_provider": "anthropic"},
|
||||
)
|
||||
assert result == "anthropic/claude-3-opus"
|
||||
|
||||
def test_existing_prefix(self):
|
||||
"""Already prefixed models unchanged."""
|
||||
result = WebFetchInterceptionLogger._resolve_full_model_name(
|
||||
"anthropic/claude-3-opus",
|
||||
{},
|
||||
)
|
||||
assert result == "anthropic/claude-3-opus"
|
||||
|
||||
def test_no_provider(self):
|
||||
"""No provider, return as-is."""
|
||||
result = WebFetchInterceptionLogger._resolve_full_model_name(
|
||||
"gpt-4",
|
||||
{},
|
||||
)
|
||||
assert result == "gpt-4"
|
||||
|
||||
|
||||
class TestCleanOptionalParams:
|
||||
"""Test _clean_optional_params."""
|
||||
|
||||
def test_basic_cleaning(self):
|
||||
"""Remove internal/chaining params."""
|
||||
params = {
|
||||
"temperature": 0.7,
|
||||
"max_tokens": 1000,
|
||||
"user": "test-user",
|
||||
}
|
||||
result = WebFetchInterceptionLogger._clean_optional_params(params)
|
||||
|
||||
assert result["temperature"] == 0.7
|
||||
assert result["max_tokens"] == 1000
|
||||
|
||||
def test_removes_internal_keys(self):
|
||||
"""Internal keys are removed."""
|
||||
params = {
|
||||
"temperature": 0.7,
|
||||
"litellm_id": "123",
|
||||
"call_type": "completion",
|
||||
}
|
||||
result = WebFetchInterceptionLogger._clean_optional_params(params)
|
||||
|
||||
assert "litellm_id" not in result
|
||||
assert "call_type" not in result
|
||||
|
||||
|
||||
class TestExtractUrlFromToolCall:
|
||||
"""Test _extract_url_from_tool_call."""
|
||||
|
||||
def test_from_input_url(self):
|
||||
"""Extract URL from input.url."""
|
||||
tool_call = {
|
||||
"input": {"url": "https://example.com"},
|
||||
}
|
||||
result = WebFetchInterceptionLogger._extract_url_from_tool_call(tool_call)
|
||||
assert result == "https://example.com"
|
||||
|
||||
def test_from_function_args(self):
|
||||
"""Extract URL from function.arguments."""
|
||||
tool_call = {
|
||||
"function": {"arguments": {"url": "https://test.com"}},
|
||||
}
|
||||
result = WebFetchInterceptionLogger._extract_url_from_tool_call(tool_call)
|
||||
assert result == "https://test.com"
|
||||
|
||||
def test_not_found(self):
|
||||
"""URL not found returns None."""
|
||||
tool_call = {"other": "data"}
|
||||
result = WebFetchInterceptionLogger._extract_url_from_tool_call(tool_call)
|
||||
assert result is None
|
||||
|
||||
|
||||
class TestCreateEmptyFetchResult:
|
||||
"""Test _create_empty_fetch_result."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_returns_string(self):
|
||||
"""Returns error string."""
|
||||
result = await WebFetchInterceptionLogger._create_empty_fetch_result()
|
||||
assert isinstance(result, str)
|
||||
assert "Failed" in result or "error" in result.lower()
|
||||
|
||||
|
||||
class TestAsyncBuildAnthropicRequestPatch:
|
||||
"""Test async_build_anthropic_request_patch."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_basic(self):
|
||||
"""Build Anthropic request patch from kwargs."""
|
||||
kwargs = {
|
||||
"model": "claude-3-opus",
|
||||
"messages": [{"role": "user", "content": "Hello"}],
|
||||
"max_tokens": 1024,
|
||||
}
|
||||
|
||||
with patch(
|
||||
"litellm.integrations.webfetch_interception.handler.WebFetchInterceptionLogger._resolve_full_model_name",
|
||||
return_value="anthropic/claude-3-opus",
|
||||
):
|
||||
result = await WebFetchInterceptionLogger.async_build_anthropic_request_patch(
|
||||
kwargs=kwargs,
|
||||
original_anthropic_messages=[],
|
||||
anthropic_headers={"Authorization": "Bearer test"},
|
||||
)
|
||||
|
||||
assert "model" in result
|
||||
assert "messages" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_with_tools(self):
|
||||
"""Patch includes tools."""
|
||||
kwargs = {
|
||||
"model": "claude-3",
|
||||
"tools": [{"type": "web_fetch_20250305", "name": "web_fetch"}],
|
||||
"messages": [],
|
||||
}
|
||||
|
||||
with patch(
|
||||
"litellm.integrations.webfetch_interception.handler.WebFetchInterceptionLogger._resolve_full_model_name",
|
||||
return_value="anthropic/claude-3",
|
||||
):
|
||||
with patch(
|
||||
"litellm.integrations.webfetch_interception.transformation.WebFetchTransformation.convert_native_to_litellm",
|
||||
return_value={"name": "litellm-web-fetch", "input_schema": {}}),
|
||||
):
|
||||
result = await WebFetchInterceptionLogger.async_build_anthropic_request_patch(
|
||||
kwargs=kwargs,
|
||||
original_anthropic_messages=[],
|
||||
anthropic_headers={},
|
||||
)
|
||||
|
||||
assert "tools" in result
|
||||
|
||||
|
||||
class TestPreCallDeploymentHook:
|
||||
"""Test async_pre_call_deployment_hook."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_no_fetch_tools_no_action(self):
|
||||
"""Without fetch_tools in router, pass through."""
|
||||
logger = WebFetchInterceptionLogger()
|
||||
kwargs = {"model": "gpt-4", "messages": []}
|
||||
|
||||
result = await logger.async_pre_call_deployment_hook(kwargs)
|
||||
assert result == kwargs
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_with_router_fetch_tools(self):
|
||||
"""With router fetch_tools, intercept tools."""
|
||||
logger = WebFetchInterceptionLogger()
|
||||
logger.router_fetch_tools = [
|
||||
{"fetch_tool_name": "fetch-1", "litellm_params": {"provider": "firecrawl"}}
|
||||
]
|
||||
|
||||
kwargs = {
|
||||
"model": "claude-3",
|
||||
"messages": [],
|
||||
"tools": [{"type": "web_fetch_20250305", "name": "web_fetch"}],
|
||||
}
|
||||
|
||||
result = await logger.async_pre_call_deployment_hook(kwargs)
|
||||
assert "tools" in result
|
||||
126
tests/webfetch_tests/test_tools_extended.py
Normal file
126
tests/webfetch_tests/test_tools_extended.py
Normal file
|
|
@ -0,0 +1,126 @@
|
|||
"""Tests for remaining uncovered methods in tools.py.
|
||||
|
||||
These tests cover functions not already covered by test_webfetch_interception.py.
|
||||
"""
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.integrations.webfetch_interception.tools import (
|
||||
get_litellm_web_fetch_tool_openai,
|
||||
is_web_fetch_tool_chat_completion,
|
||||
is_native_fetch_tool,
|
||||
convert_native_fetch_to_litellm,
|
||||
LITELLM_NATIVE_FETCH_TOOLS,
|
||||
LITELLM_WEB_FETCH_TOOL_NAME,
|
||||
)
|
||||
|
||||
|
||||
class TestGetLitellmWebFetchToolOpenAI:
|
||||
"""Test OpenAI format tool definition."""
|
||||
|
||||
def test_returns_correct_name(self):
|
||||
"""Tool name matches constant."""
|
||||
tool = get_litellm_web_fetch_tool_openai()
|
||||
assert tool["type"] == "function"
|
||||
assert tool["function"]["name"] == LITELLM_WEB_FETCH_TOOL_NAME
|
||||
|
||||
def test_has_url_property(self):
|
||||
"""Schema requires url parameter."""
|
||||
tool = get_litellm_web_fetch_tool_openai()
|
||||
assert "url" in tool["function"]["parameters"]["properties"]
|
||||
assert "required" in tool["function"]["parameters"]
|
||||
|
||||
|
||||
class TestIsNativeFetchTool:
|
||||
"""Test native fetch tool detection."""
|
||||
|
||||
def test_web_fetch_20250305(self):
|
||||
"""Anthropic format is native."""
|
||||
assert is_native_fetch_tool("web_fetch_20250305") is True
|
||||
|
||||
def test_web_fetch_name(self):
|
||||
"""web_fetch name is native."""
|
||||
assert is_native_fetch_tool("web_fetch") is True
|
||||
|
||||
def test_WebFetch(self):
|
||||
"""Legacy WebFetch is native."""
|
||||
assert is_native_fetch_tool("WebFetch") is True
|
||||
|
||||
def test_non_native(self):
|
||||
"""Other tools are not native."""
|
||||
assert is_native_fetch_tool("calculator") is False
|
||||
assert is_native_fetch_tool("") is False
|
||||
|
||||
def test_type_matching(self):
|
||||
"""Tool type matching works."""
|
||||
assert is_native_fetch_tool("anything", "web_fetch_20250305") is True
|
||||
assert is_native_fetch_tool("anything", "other_type") is False
|
||||
|
||||
|
||||
class TestIsWebFetchToolChatCompletion:
|
||||
"""Test chat completion tool detection."""
|
||||
|
||||
def test_litellm_name(self):
|
||||
"""LiteLLM tool name is detected."""
|
||||
tool = {"function": {"name": LITELLM_WEB_FETCH_TOOL_NAME}}
|
||||
assert is_web_fetch_tool_chat_completion(tool) is True
|
||||
|
||||
def test_anthropic_type(self):
|
||||
"""Anthropic type tool."""
|
||||
tool = {"type": "web_fetch_20250305"}
|
||||
assert is_web_fetch_tool_chat_completion(tool) is True
|
||||
|
||||
def test_non_fetch(self):
|
||||
"""Non-fetch tool not detected."""
|
||||
assert is_web_fetch_tool_chat_completion({"function": {"name": "other"}}) is False
|
||||
|
||||
|
||||
class TestConvertNativeFetchToLitellm:
|
||||
"""Test native to LiteLLM format conversion."""
|
||||
|
||||
def test_basic_conversion(self):
|
||||
"""Basic tool conversion."""
|
||||
native = {
|
||||
"type": "web_fetch_20250305",
|
||||
"name": "web_fetch",
|
||||
"input_schema": {
|
||||
"properties": {"url": {"type": "string"}},
|
||||
"required": ["url"],
|
||||
},
|
||||
}
|
||||
|
||||
result = convert_native_fetch_to_litellm(native)
|
||||
|
||||
assert result["name"] == LITELLM_WEB_FETCH_TOOL_NAME
|
||||
assert "url" in result["input_schema"]["properties"]
|
||||
|
||||
def test_preserves_description(self):
|
||||
"""Description is preserved if present."""
|
||||
native = {
|
||||
"description": "Custom fetch",
|
||||
"input_schema": {},
|
||||
}
|
||||
|
||||
result = convert_native_fetch_to_litellm(native)
|
||||
assert result.get("description") == "Custom fetch"
|
||||
|
||||
def test_preserves_required(self):
|
||||
"""Required fields preserved."""
|
||||
native = {
|
||||
"input_schema": {
|
||||
"required": ["url", "extra"],
|
||||
},
|
||||
}
|
||||
|
||||
result = convert_native_fetch_to_litellm(native)
|
||||
assert "extra" in result["input_schema"]["required"]
|
||||
|
||||
def test_empty_input_schema(self):
|
||||
"""Empty input schema handled gracefully."""
|
||||
native = {
|
||||
"type": "web_fetch_20250305",
|
||||
}
|
||||
|
||||
result = convert_native_fetch_to_litellm(native)
|
||||
assert result["name"] == LITELLM_WEB_FETCH_TOOL_NAME
|
||||
assert result["input_schema"]["properties"]["url"]["type"] == "string"
|
||||
Loading…
Add table
Reference in a new issue