mirror of
https://github.com/BerriAI/litellm.git
synced 2026-08-28 05:25:59 +00:00
9c014716ec
Search providers resolved the server-configured API key (e.g.
get_secret_str("SERPER_API_KEY")) in validate_environment whenever the
caller omitted api_key, while get_complete_url independently honored a
caller-supplied api_base. A caller who passes their own api_base and no
api_key therefore made the proxy send the operator's provider key to a
host they control; POST /search_tools/test_connection forwards
request-body api_base/api_key straight into asearch, so any authenticated
user could exfiltrate the server's search credentials.
Add a shared host-aware fallback in BaseSearchConfig.resolve_server_api_key
that only applies a server-managed secret when the caller-supplied
api_base is absent or resolves to a trusted host (the provider default or
the operator's own *_API_BASE env override); otherwise it refuses and asks
for an explicit api_key. The guard only triggers when a server secret
actually exists, so keyless and self-hosted providers (searxng, you.com
free tier) keep working. Every provider that carries a server secret is
migrated to the helper; dataforseo reuses the same guard for its
login:password basic-auth credentials.
This changes behavior for callers that previously passed a per-request
api_base while relying on a server-configured key: they must now pass an
explicit api_key, or the operator must configure the base via the
provider's *_API_BASE env var (which stays trusted).
292 lines
11 KiB
Python
292 lines
11 KiB
Python
"""
|
|
Tests for APISerpent search API integration (quick + deep search).
|
|
"""
|
|
|
|
import os
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from urllib.parse import parse_qs, urlparse
|
|
|
|
import pytest
|
|
|
|
import litellm
|
|
from litellm.llms.apiserpent.search.defaults import APISerpentSearchParams
|
|
from litellm.llms.apiserpent.search.transformation import APISerpentSearchConfig
|
|
from litellm.llms.base_llm.search.transformation import SearchResponse
|
|
|
|
|
|
def _params(config, query, optional_params):
|
|
return config.transform_search_request(
|
|
query=query, optional_params=optional_params
|
|
)["_apiserpent_params"]
|
|
|
|
|
|
class TestAPISerpentDefaults:
|
|
def test_defaults_applied(self):
|
|
params = APISerpentSearchParams().to_request_params()
|
|
assert params["engine"] == "google"
|
|
assert params["country"] == "us"
|
|
assert params["num"] == 10
|
|
assert params["format"] == "full"
|
|
assert "freshness" not in params
|
|
assert "pixel_position" not in params
|
|
|
|
def test_bool_lowercased(self):
|
|
params = APISerpentSearchParams(pixel_position=True).to_request_params()
|
|
assert params["pixel_position"] == "true"
|
|
|
|
@pytest.mark.parametrize("num", [0, 101, 500])
|
|
def test_num_out_of_range_raises(self, num):
|
|
with pytest.raises(ValueError, match="num must be between 1 and 100"):
|
|
APISerpentSearchParams(num=num)
|
|
|
|
@pytest.mark.parametrize("pages", [0, 11, 50])
|
|
def test_pages_out_of_range_raises(self, pages):
|
|
with pytest.raises(ValueError, match="pages must be between 1 and 10"):
|
|
APISerpentSearchParams(pages=pages)
|
|
|
|
def test_valid_bounds_accepted(self):
|
|
params = APISerpentSearchParams(num=100, pages=10).to_request_params()
|
|
assert params["num"] == 100
|
|
assert params["pages"] == 10
|
|
|
|
|
|
class TestAPISerpentConfig:
|
|
def test_ui_friendly_name(self):
|
|
assert APISerpentSearchConfig().ui_friendly_name() == "APISerpent"
|
|
|
|
def test_get_http_method(self):
|
|
assert APISerpentSearchConfig().get_http_method() == "GET"
|
|
|
|
@patch("litellm.llms.apiserpent.search.transformation.get_secret_str")
|
|
def test_validate_environment_with_api_key(self, mock_get_secret):
|
|
mock_get_secret.return_value = None
|
|
headers = APISerpentSearchConfig().validate_environment(
|
|
{}, api_key="test-api-key"
|
|
)
|
|
assert headers["X-API-Key"] == "test-api-key"
|
|
assert headers["Content-Type"] == "application/json"
|
|
|
|
def test_validate_environment_without_api_key(self, monkeypatch):
|
|
monkeypatch.delenv("APISERPENT_API_KEY", raising=False)
|
|
with pytest.raises(ValueError, match="APISERPENT_API_KEY is not set"):
|
|
APISerpentSearchConfig().validate_environment({})
|
|
|
|
def test_transform_request_basic_applies_defaults(self):
|
|
params = _params(APISerpentSearchConfig(), "test query", {})
|
|
assert params["q"] == "test query"
|
|
assert params["engine"] == "google"
|
|
assert params["num"] == 10
|
|
|
|
def test_transform_request_list_query_joined(self):
|
|
assert _params(APISerpentSearchConfig(), ["foo", "bar"], {})["q"] == "foo bar"
|
|
|
|
def test_quick_num_clamped(self):
|
|
config = APISerpentSearchConfig()
|
|
assert _params(config, "q", {"max_results": 250})["num"] == 100
|
|
assert _params(config, "q", {"max_results": 0})["num"] == 1
|
|
|
|
def test_deep_num_floor_is_10(self):
|
|
config = APISerpentSearchConfig()
|
|
params = _params(config, "q", {"deep": True, "max_results": 5})
|
|
assert params["num"] == 10
|
|
|
|
def test_country_lowercased(self):
|
|
assert (
|
|
_params(APISerpentSearchConfig(), "q", {"country": "US"})["country"] == "us"
|
|
)
|
|
|
|
def test_engine_and_optional_passthrough(self):
|
|
params = _params(
|
|
APISerpentSearchConfig(),
|
|
"q",
|
|
{"engine": "bing", "language": "es", "freshness": "d", "safe": "strict"},
|
|
)
|
|
assert params["engine"] == "bing"
|
|
assert params["language"] == "es"
|
|
assert params["freshness"] == "d"
|
|
assert params["safe"] == "strict"
|
|
|
|
def test_pixel_position_passthrough_lowercased(self):
|
|
params = _params(APISerpentSearchConfig(), "q", {"pixel_position": True})
|
|
assert params["pixel_position"] == "true"
|
|
|
|
def test_domain_filter(self):
|
|
params = _params(
|
|
APISerpentSearchConfig(),
|
|
"machine learning",
|
|
{"search_domain_filter": ["arxiv.org", "nature.com"]},
|
|
)
|
|
assert "site:arxiv.org" in params["q"]
|
|
assert "site:nature.com" in params["q"]
|
|
assert "machine learning" in params["q"]
|
|
|
|
def test_get_complete_url_quick_path(self):
|
|
config = APISerpentSearchConfig()
|
|
data = {"_apiserpent_params": {"q": "test", "num": 5}}
|
|
url = config.get_complete_url(api_base=None, optional_params={}, data=data)
|
|
parsed = urlparse(url)
|
|
assert (
|
|
f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
|
|
== "https://apiserpent.com/api/search/quick"
|
|
)
|
|
assert parse_qs(parsed.query)["q"] == ["test"]
|
|
|
|
def test_get_complete_url_deep_path(self):
|
|
config = APISerpentSearchConfig()
|
|
data = {"_apiserpent_params": {"q": "test"}}
|
|
url = config.get_complete_url(
|
|
api_base=None, optional_params={"deep": True}, data=data
|
|
)
|
|
parsed = urlparse(url)
|
|
assert (
|
|
f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
|
|
== "https://apiserpent.com/api/search"
|
|
)
|
|
|
|
def test_explicit_api_base_swaps_host_and_keeps_routing(self):
|
|
config = APISerpentSearchConfig()
|
|
url = config.get_complete_url(
|
|
api_base="https://staging.apiserpent.com",
|
|
optional_params={"deep": True},
|
|
data={"_apiserpent_params": {"q": "x"}},
|
|
)
|
|
parsed = urlparse(url)
|
|
assert (
|
|
f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
|
|
== "https://staging.apiserpent.com/api/search"
|
|
)
|
|
|
|
def test_get_complete_url_is_idempotent(self):
|
|
"""The handler re-invokes get_complete_url with the resolved URL as api_base."""
|
|
config = APISerpentSearchConfig()
|
|
resolved = config.get_complete_url(
|
|
api_base=None, optional_params={"deep": True}, data=None
|
|
)
|
|
again = config.get_complete_url(
|
|
api_base=resolved,
|
|
optional_params={"deep": True},
|
|
data={"_apiserpent_params": {"q": "x"}},
|
|
)
|
|
assert again == "https://apiserpent.com/api/search?q=x"
|
|
assert "/api/search/api/search" not in again
|
|
|
|
def test_transform_response_full_format(self):
|
|
raw_response = MagicMock()
|
|
raw_response.json.return_value = {
|
|
"success": True,
|
|
"results": {
|
|
"organic": [
|
|
{"title": "R1", "url": "https://example.com/1", "snippet": "S1"},
|
|
{"title": "R2", "url": "https://example.com/2", "snippet": "S2"},
|
|
]
|
|
},
|
|
}
|
|
response = APISerpentSearchConfig().transform_search_response(
|
|
raw_response=raw_response, logging_obj=None
|
|
)
|
|
assert isinstance(response, SearchResponse)
|
|
assert len(response.results) == 2
|
|
assert response.results[0].title == "R1"
|
|
assert response.results[0].url == "https://example.com/1"
|
|
|
|
def test_transform_response_simple_format(self):
|
|
raw_response = MagicMock()
|
|
raw_response.json.return_value = {
|
|
"success": True,
|
|
"results": [{"position": 1, "title": "R1", "url": "https://example.com/1"}],
|
|
}
|
|
response = APISerpentSearchConfig().transform_search_response(
|
|
raw_response=raw_response, logging_obj=None
|
|
)
|
|
assert len(response.results) == 1
|
|
assert response.results[0].title == "R1"
|
|
|
|
def test_transform_response_empty(self):
|
|
raw_response = MagicMock()
|
|
raw_response.json.return_value = {"success": True, "results": {}}
|
|
response = APISerpentSearchConfig().transform_search_response(
|
|
raw_response=raw_response, logging_obj=None
|
|
)
|
|
assert len(response.results) == 0
|
|
|
|
def test_transform_response_null_results(self):
|
|
"""An error response with `results: null` must not raise."""
|
|
raw_response = MagicMock()
|
|
raw_response.json.return_value = {"success": False, "results": None}
|
|
response = APISerpentSearchConfig().transform_search_response(
|
|
raw_response=raw_response, logging_obj=None
|
|
)
|
|
assert response.results == []
|
|
|
|
|
|
class TestAPISerpentSearchIntegration:
|
|
@staticmethod
|
|
def _mock_response():
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {
|
|
"success": True,
|
|
"results": {
|
|
"organic": [
|
|
{
|
|
"title": "Test Result",
|
|
"url": "https://example.com",
|
|
"snippet": "A snippet",
|
|
}
|
|
]
|
|
},
|
|
}
|
|
return mock_response
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_asearch_quick_default(self):
|
|
os.environ["APISERPENT_API_KEY"] = "test-api-key"
|
|
with patch(
|
|
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.get",
|
|
new_callable=AsyncMock,
|
|
) as mock_get:
|
|
mock_get.return_value = self._mock_response()
|
|
|
|
response = await litellm.asearch(
|
|
query="latest developments in AI",
|
|
search_provider="apiserpent",
|
|
max_results=5,
|
|
country="US",
|
|
)
|
|
|
|
parsed = urlparse(mock_get.call_args.kwargs["url"])
|
|
assert (
|
|
f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
|
|
== "https://apiserpent.com/api/search/quick"
|
|
)
|
|
qs = parse_qs(parsed.query)
|
|
assert qs["q"] == ["latest developments in AI"]
|
|
assert qs["num"] == ["5"]
|
|
assert qs["country"] == ["us"]
|
|
assert mock_get.call_args.kwargs["headers"]["X-API-Key"] == "test-api-key"
|
|
|
|
assert response.object == "search"
|
|
assert response.results[0].title == "Test Result"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_asearch_deep(self):
|
|
os.environ["APISERPENT_API_KEY"] = "test-api-key"
|
|
with patch(
|
|
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.get",
|
|
new_callable=AsyncMock,
|
|
) as mock_get:
|
|
mock_get.return_value = self._mock_response()
|
|
|
|
await litellm.asearch(
|
|
query="climate research",
|
|
search_provider="apiserpent",
|
|
deep=True,
|
|
max_results=40,
|
|
)
|
|
|
|
parsed = urlparse(mock_get.call_args.kwargs["url"])
|
|
assert (
|
|
f"{parsed.scheme}://{parsed.netloc}{parsed.path}"
|
|
== "https://apiserpent.com/api/search"
|
|
)
|
|
assert parse_qs(parsed.query)["num"] == ["40"]
|