diff --git a/litellm/llms/scavio/__init__.py b/litellm/llms/scavio/__init__.py new file mode 100644 index 00000000000..9f8c874dae5 --- /dev/null +++ b/litellm/llms/scavio/__init__.py @@ -0,0 +1,3 @@ +from litellm.llms.scavio.search.transformation import ScavioSearchConfig + +__all__ = ("ScavioSearchConfig",) diff --git a/litellm/llms/scavio/search/__init__.py b/litellm/llms/scavio/search/__init__.py new file mode 100644 index 00000000000..9f8c874dae5 --- /dev/null +++ b/litellm/llms/scavio/search/__init__.py @@ -0,0 +1,3 @@ +from litellm.llms.scavio.search.transformation import ScavioSearchConfig + +__all__ = ("ScavioSearchConfig",) diff --git a/litellm/llms/scavio/search/transformation.py b/litellm/llms/scavio/search/transformation.py new file mode 100644 index 00000000000..b6ec1ebd332 --- /dev/null +++ b/litellm/llms/scavio/search/transformation.py @@ -0,0 +1,209 @@ +""" +Calls Scavio's /api/v2/google endpoint to search Google. + +Scavio API Reference: https://scavio.dev/docs/search-api +""" + +from __future__ import annotations + +from collections.abc import Mapping +from types import MappingProxyType +from typing import TYPE_CHECKING, Final + +import httpx +from pydantic import ConfigDict, StrictInt, TypeAdapter, ValidationError + +from litellm.llms.base_llm.chat.transformation import BaseLLMException +from litellm.llms.base_llm.search.transformation import ( + BaseSearchConfig, + SearchResponse, + SearchResult, +) +from litellm.types.llms.base import LiteLLMBaseModel + +if TYPE_CHECKING: + from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj + +_SCAVIO_DOCS_URL: Final = "https://scavio.dev/docs/search-api" +_SEARCH_PATH: Final = "/api/v2/google" + + +class _ScavioOrganicResult(LiteLLMBaseModel): + model_config = ConfigDict(extra="ignore", frozen=True) + + title: str | None = None + link: str | None = None + snippet: str | None = None + + +class _ScavioSearchResponse(LiteLLMBaseModel): + model_config = ConfigDict(extra="ignore", frozen=True) + + organic_results: tuple[_ScavioOrganicResult, ...] + + +class _ErrorEnvelope(LiteLLMBaseModel): + model_config = ConfigDict(extra="ignore", frozen=True) + + error: str | None = None + + +class _UnifiedParams(LiteLLMBaseModel): + model_config = ConfigDict(extra="ignore", frozen=True) + + max_results: StrictInt | None = None + + +_DomainListAdapter: Final = TypeAdapter(tuple[str, ...]) + +_NOTHING: Final[Mapping[str, object]] = MappingProxyType({}) + + +def _optional(key: str, value: object) -> Mapping[str, object]: + return MappingProxyType({key: value}) if value is not None else _NOTHING + + +class ScavioSearchConfig(BaseSearchConfig): + SCAVIO_API_BASE = "https://api.scavio.dev" + + @staticmethod + def ui_friendly_name() -> str: + return "Scavio" + + def validate_environment( + self, + headers: dict[str, str], # mutable-ok: BaseSearchConfig.validate_environment signature + api_key: str | None = None, + api_base: str | None = None, + **kwargs: object, # kwargs-ok: BaseSearchConfig.validate_environment signature + ) -> dict[str, str]: # mutable-ok: the http handler passes this straight to httpx as headers + resolved_api_key: Final = self.resolve_server_api_key( + caller_api_key=api_key, + caller_api_base=api_base, + key_env_vars=("SCAVIO_API_KEY",), + base_env_var=None, + default_api_base=self.SCAVIO_API_BASE, + ) + if not resolved_api_key: + raise ValueError("SCAVIO_API_KEY is not set. Set `SCAVIO_API_KEY` environment variable.") + return { + **headers, + "Authorization": f"Bearer {resolved_api_key}", + "Content-Type": "application/json", + } + + def get_complete_url( + self, + api_base: str | None, + optional_params: dict[str, object], # mutable-ok: BaseSearchConfig.get_complete_url signature + data: dict[str, object] | list[dict[str, object]] | None = None, # mutable-ok: base signature + **kwargs: object, # kwargs-ok: BaseSearchConfig.get_complete_url signature + ) -> str: + resolved_base: Final = (api_base or self.SCAVIO_API_BASE).rstrip("/") + if resolved_base.endswith(_SEARCH_PATH): + return resolved_base + return f"{resolved_base}{_SEARCH_PATH}" + + def transform_search_request( + self, + query: str | list[str], # mutable-ok: BaseSearchConfig.transform_search_request signature + optional_params: dict[str, object], # mutable-ok: base signature + **kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_request signature + ) -> dict[str, object]: # mutable-ok: the http handler passes this straight to httpx as the JSON body + """ + Scavio returns one Google page per call and has no result-count param, so `max_results` is + applied to the parsed response instead. `resolve_ai_overview` defaults to false because only + organic results are mapped and resolving a deferred AI Overview costs an extra upstream round trip + """ + unified_params: Final = self.get_supported_perplexity_optional_params() + country: Final = optional_params.get("country") + joined_query: Final = " ".join(query) if isinstance(query, list) else query + + passthrough: Final = MappingProxyType( + {param: value for param, value in optional_params.items() if param not in unified_params} + ) + + return { + "resolve_ai_overview": False, + **_optional("gl", country.lower() if isinstance(country, str) else None), + **passthrough, + "query": _with_domain_filter(joined_query, optional_params.get("search_domain_filter")), + } + + def transform_search_response( + self, + raw_response: httpx.Response, + logging_obj: LiteLLMLoggingObj, + **kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_response signature + ) -> SearchResponse: + try: + parsed: Final = _ScavioSearchResponse.model_validate_json(raw_response.content) + except ValidationError as e: + raise self.get_error_class( + error_message=f"response does not match the documented /api/v2/google schema: {e}", + status_code=raw_response.status_code, + headers=dict(raw_response.headers), + ) + + max_results: Final = _requested_max_results(kwargs.get("optional_params")) + organic: Final = parsed.organic_results[:max_results] if max_results is not None else parsed.organic_results + + return SearchResponse( + results=[ + SearchResult( + title=result.title or "", + url=result.link or "", + snippet=result.snippet or "", + date=None, + last_updated=None, + ) + for result in organic + ], + object="search", + ) + + def get_error_class( + self, + error_message: str, + status_code: int, + headers: dict[str, str], # mutable-ok: BaseSearchConfig.get_error_class signature + ) -> Exception: + detail: Final = _unwrap_error_detail(error_message).rstrip(". ") + return BaseLLMException( + status_code=status_code, + message=f"Scavio Search: {detail}. See {_SCAVIO_DOCS_URL} for details.", + headers=headers, + ) + + +def _unwrap_error_detail(error_message: str) -> str: + try: + body: Final = _ErrorEnvelope.model_validate_json(error_message) + except ValidationError: + return error_message + return body.error or error_message + + +def _with_domain_filter(query: str, search_domain_filter: object) -> str: + """ + Google ignores the `site:` filter on `(query) (site:a)` and on an ungrouped `site:a OR site:b`, + so the query stays bare, one include is `site:a`, and two or more are grouped as `(site:a OR site:b)` + """ + try: + domains: Final = _DomainListAdapter.validate_python(search_domain_filter) + except ValidationError: + return query + included: Final = tuple(d for d in domains if d and not d.startswith("-")) + excluded: Final = tuple(d[1:] for d in domains if d.startswith("-") and len(d) > 1) + include_sites: Final = tuple(f"site:{d}" for d in included) + include_clause: Final = f"({' OR '.join(include_sites)})" if len(include_sites) > 1 else "".join(include_sites) + operators: Final = tuple(clause for clause in (include_clause, *(f"-site:{d}" for d in excluded)) if clause) + return " ".join((query, *operators)) + + +def _requested_max_results(optional_params: object) -> int | None: + try: + max_results: Final = _UnifiedParams.model_validate(optional_params).max_results + except ValidationError: + return None + return max_results if max_results is not None and max_results > 0 else None diff --git a/litellm/model_prices_and_context_window_backup.json b/litellm/model_prices_and_context_window_backup.json index 663c6d1a2a7..d6f494552ae 100644 --- a/litellm/model_prices_and_context_window_backup.json +++ b/litellm/model_prices_and_context_window_backup.json @@ -23226,6 +23226,14 @@ "notes": "Nimble Search API pay-as-you-go list price: $5 per 1,000 searches, up to 100 results per search. Volume plans price differently." } }, + "scavio/search": { + "input_cost_per_query": 0.01, + "litellm_provider": "scavio", + "mode": "search", + "metadata": { + "notes": "Scavio Google Search API pay-as-you-go list price: $0.01 per search (1 credit). Subscription plans price lower per credit." + } + }, "elevenlabs/scribe_v1": { "input_cost_per_second": 6.11e-05, "litellm_provider": "elevenlabs", diff --git a/litellm/types/utils.py b/litellm/types/utils.py index a99c22e9479..2e7e78ba48b 100644 --- a/litellm/types/utils.py +++ b/litellm/types/utils.py @@ -4322,6 +4322,7 @@ class SearchProviders(str, Enum): TINYFISH = "tinyfish" AGENTCORE = "agentcore" NIMBLE = "nimble" + SCAVIO = "scavio" BING_GROUNDING = "bing_grounding" diff --git a/litellm/utils.py b/litellm/utils.py index 7a334f8974b..c9d8e5e9bb3 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -9933,6 +9933,7 @@ class ProviderConfigManager: ParallelAISearchConfig, ) from litellm.llms.perplexity.search.transformation import PerplexitySearchConfig + from litellm.llms.scavio.search.transformation import ScavioSearchConfig from litellm.llms.searchapi.search.transformation import SearchAPIConfig from litellm.llms.searxng.search.transformation import SearXNGSearchConfig from litellm.llms.serper.search.transformation import SerperSearchConfig @@ -9960,6 +9961,7 @@ class ProviderConfigManager: SearchProviders.TINYFISH: TinyfishSearchConfig, SearchProviders.AGENTCORE: AgentCoreSearchConfig, SearchProviders.NIMBLE: NimbleSearchConfig, + SearchProviders.SCAVIO: ScavioSearchConfig, SearchProviders.BING_GROUNDING: BingGroundingSearchConfig, } config_class: Final = PROVIDER_TO_CONFIG_MAP.get(provider, None) diff --git a/model_prices_and_context_window.json b/model_prices_and_context_window.json index 663c6d1a2a7..d6f494552ae 100644 --- a/model_prices_and_context_window.json +++ b/model_prices_and_context_window.json @@ -23226,6 +23226,14 @@ "notes": "Nimble Search API pay-as-you-go list price: $5 per 1,000 searches, up to 100 results per search. Volume plans price differently." } }, + "scavio/search": { + "input_cost_per_query": 0.01, + "litellm_provider": "scavio", + "mode": "search", + "metadata": { + "notes": "Scavio Google Search API pay-as-you-go list price: $0.01 per search (1 credit). Subscription plans price lower per credit." + } + }, "elevenlabs/scribe_v1": { "input_cost_per_second": 6.11e-05, "litellm_provider": "elevenlabs", diff --git a/provider_endpoints_support.json b/provider_endpoints_support.json index 62e23388581..1f794571e1f 100644 --- a/provider_endpoints_support.json +++ b/provider_endpoints_support.json @@ -2619,6 +2619,13 @@ "search": true } }, + "scavio": { + "display_name": "Scavio (`scavio`)", + "url": "https://scavio.dev/docs/search-api", + "endpoints": { + "search": true + } + }, "triton": { "display_name": "Triton (`triton`)", "url": "https://docs.litellm.ai/docs/providers/triton-inference-server", diff --git a/tests/code_coverage_tests/enforce_llms_folder_style.py b/tests/code_coverage_tests/enforce_llms_folder_style.py index a284cf9e1a9..c6277a6cf5f 100644 --- a/tests/code_coverage_tests/enforce_llms_folder_style.py +++ b/tests/code_coverage_tests/enforce_llms_folder_style.py @@ -21,6 +21,7 @@ SEARCH_PROVIDERS = [ "apiserpent", "tinyfish", "nimble", + "scavio", ] ALLOWED_FILES_IN_LLMS_FOLDER = [ diff --git a/tests/unit/llms/base_llm/search/test_base_search_transformation.py b/tests/unit/llms/base_llm/search/test_base_search_transformation.py index e6aad7688d1..017792dcd1b 100644 --- a/tests/unit/llms/base_llm/search/test_base_search_transformation.py +++ b/tests/unit/llms/base_llm/search/test_base_search_transformation.py @@ -31,6 +31,7 @@ from litellm.llms.linkup.search.transformation import LinkupSearchConfig from litellm.llms.nimble.search.transformation import NimbleSearchConfig from litellm.llms.parallel_ai.search.transformation import ParallelAISearchConfig from litellm.llms.perplexity.search.transformation import PerplexitySearchConfig +from litellm.llms.scavio.search.transformation import ScavioSearchConfig from litellm.llms.searchapi.search.transformation import SearchAPIConfig from litellm.llms.searxng.search.transformation import SearXNGSearchConfig from litellm.llms.serper.search.transformation import SerperSearchConfig @@ -101,6 +102,7 @@ PROVIDERS: Tuple[ProviderSpec, ...] = ( (TinyfishSearchConfig, {"TINYFISH_API_KEY": "srv"}, "caller-key", {}), (FastCRWSearchConfig, {"CRW_API_KEY": "srv"}, "caller-key", {}), (NimbleSearchConfig, {"NIMBLE_API_KEY": "srv"}, "caller-key", {}), + (ScavioSearchConfig, {"SCAVIO_API_KEY": "srv"}, "caller-key", {}), (BingGroundingSearchConfig, {"BING_GROUNDING_TOKEN": "srv"}, "caller-key", {}), ) diff --git a/tests/unit/llms/scavio/__init__.py b/tests/unit/llms/scavio/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/llms/scavio/search/__init__.py b/tests/unit/llms/scavio/search/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/llms/scavio/search/test_scavio_search_transformation.py b/tests/unit/llms/scavio/search/test_scavio_search_transformation.py new file mode 100644 index 00000000000..9981f4c41fd --- /dev/null +++ b/tests/unit/llms/scavio/search/test_scavio_search_transformation.py @@ -0,0 +1,361 @@ +import datetime +import json +from collections.abc import Mapping, Sequence +from typing import Final + +import httpx +import pytest + +import litellm +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.llms.base_llm.search.transformation import SearchResponse +from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler +from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler +from litellm.llms.scavio.search.transformation import ScavioSearchConfig +from litellm.search.cost_calculator import search_provider_cost_per_query +from litellm.types.utils import SearchProviders +from litellm.utils import ProviderConfigManager + +SCAVIO_URL: Final = "https://api.scavio.dev/api/v2/google" + +LIVE_RESPONSE: Final[Mapping[str, object]] = { + "search_parameters": {"q": "open source llm gateway", "hl": "en", "gl": "us", "device": "desktop", "start": 0}, + "search_information": {"query_displayed": "open source llm gateway", "total_results": 2}, + "organic_results": [ + { + "position": 1, + "title": "LiteLLM - GitHub", + "link": "https://github.com/BerriAI/litellm", + "displayed_link": "https://github.com > BerriAI > litellm", + "snippet": "Python SDK, Proxy Server (AI Gateway) to call 100+ LLM APIs in OpenAI format.", + "snippet_highlighted_words": ["AI Gateway"], + "source": "github.com", + }, + { + "position": 2, + "title": "Is there an open-source LLM gateway with runtime routing?", + "link": "https://www.reddit.com/r/SelfHostedAI/comments/example", + "source": "reddit.com", + }, + ], + "related_searches": [{"query": "llm gateway comparison"}], + "cached": False, + "response_time": 3.1, + "credits_used": 1, + "credits_remaining": 99, +} + + +def _config() -> ScavioSearchConfig: + return ScavioSearchConfig() + + +def _logging_obj() -> Logging: + return Logging( + model="scavio/search", + messages=[], + stream=False, + call_type="search", + start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc), + litellm_call_id="test-call", + function_id="test-function", + ) + + +def _resp(payload: Mapping[str, object] | str, status_code: int = 200) -> httpx.Response: + body: Final = payload if isinstance(payload, str) else json.dumps(payload) + return httpx.Response(status_code, content=body.encode(), request=httpx.Request("POST", SCAVIO_URL)) + + +def _result(title: str = "Test Title", link: str = "https://example.com") -> Mapping[str, object]: + return { + "position": 1, + "title": title, + "link": link, + "displayed_link": f"{link} > docs", + "snippet": "Test snippet", + "source": "example.com", + } + + +class _ScavioStub: + def __init__(self, status_code: int, body: Mapping[str, object]) -> None: + self.status_code: Final = status_code + self.body: Final = body + self.requests: tuple[httpx.Request, ...] = () + + def __call__(self, request: httpx.Request) -> httpx.Response: + self.requests = (*self.requests, request) + return httpx.Response(self.status_code, content=json.dumps(self.body).encode()) + + +def test_ui_friendly_name() -> None: + assert _config().ui_friendly_name() == "Scavio" + + +def test_search_provider_resolves_to_scavio_config() -> None: + config: Final = ProviderConfigManager.get_provider_search_config(provider=SearchProviders("scavio")) + assert isinstance(config, ScavioSearchConfig) + + +def test_validate_environment_with_explicit_key() -> None: + headers: Final = _config().validate_environment({}, api_key="explicit-key") + assert headers["Authorization"] == "Bearer explicit-key" + assert headers["Content-Type"] == "application/json" + + +def test_validate_environment_reads_env_key(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SCAVIO_API_KEY", "env-key") + assert _config().validate_environment({})["Authorization"] == "Bearer env-key" + + +def test_validate_environment_missing_key_raises(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delenv("SCAVIO_API_KEY", raising=False) + with pytest.raises(ValueError, match="SCAVIO_API_KEY"): + _config().validate_environment({}) + + +def test_validate_environment_does_not_mutate_and_is_idempotent() -> None: + config: Final = _config() + caller_headers: Final = {"X-Custom": "keep-me"} + + once: Final = config.validate_environment(caller_headers, api_key="k") + twice: Final = config.validate_environment(once, api_key="k") + + assert caller_headers == {"X-Custom": "keep-me"} + assert once == twice + assert once["X-Custom"] == "keep-me" + + +def test_get_complete_url_default_base() -> None: + assert _config().get_complete_url(None, {}) == SCAVIO_URL + + +@pytest.mark.parametrize( + "api_base", + [ + "https://proxy.local", + "https://proxy.local/", + "https://proxy.local/api/v2/google", + "https://proxy.local/api/v2/google/", + ], +) +def test_get_complete_url_appends_path_exactly_once(api_base: str) -> None: + assert _config().get_complete_url(api_base, {}) == "https://proxy.local/api/v2/google" + + +def test_transform_search_request_joins_list_query() -> None: + assert _config().transform_search_request(["foo", "bar"], {})["query"] == "foo bar" + + +def test_transform_search_request_maps_country_to_lowercase_gl() -> None: + assert _config().transform_search_request("q", {"country": "GB"})["gl"] == "gb" + + +def test_transform_search_request_explicit_gl_wins_over_country() -> None: + assert _config().transform_search_request("q", {"country": "GB", "gl": "de"})["gl"] == "de" + + +@pytest.mark.parametrize("unified_param", ["max_results", "max_tokens_per_page", "country", "search_domain_filter"]) +def test_transform_search_request_does_not_leak_unified_params(unified_param: str) -> None: + data: Final = _config().transform_search_request( + "q", + {"max_results": 3, "max_tokens_per_page": 1024, "country": "us", "search_domain_filter": ["a.com"]}, + ) + assert unified_param not in data + + +def test_transform_search_request_skips_ai_overview_resolution_by_default() -> None: + assert _config().transform_search_request("q", {})["resolve_ai_overview"] is False + + +def test_transform_search_request_caller_can_enable_ai_overview_resolution() -> None: + assert _config().transform_search_request("q", {"resolve_ai_overview": True})["resolve_ai_overview"] is True + + +@pytest.mark.parametrize( + "domains, expected", + [ + (["arxiv.org"], "q site:arxiv.org"), + (["arxiv.org", "nature.com"], "q (site:arxiv.org OR site:nature.com)"), + (["arxiv.org", "nature.com", "science.org"], "q (site:arxiv.org OR site:nature.com OR site:science.org)"), + (["-spam.com"], "q -site:spam.com"), + (["arxiv.org", "-spam.com", "-junk.net"], "q site:arxiv.org -site:spam.com -site:junk.net"), + (["-spam.com", "arxiv.org", "nature.com"], "q (site:arxiv.org OR site:nature.com) -site:spam.com"), + ([], "q"), + (["", "-"], "q"), + ("arxiv.org", "q"), + ], +) +def test_transform_search_request_folds_domain_filter_into_query(domains: Sequence[str], expected: str) -> None: + assert _config().transform_search_request("q", {"search_domain_filter": domains})["query"] == expected + + +def test_transform_search_request_without_domain_filter_leaves_query_unchanged() -> None: + assert _config().transform_search_request("latest AI developments", {})["query"] == "latest AI developments" + + +def test_transform_search_request_never_wraps_user_query_in_parentheses() -> None: + data: Final = _config().transform_search_request( + ["latest", "AI developments"], {"search_domain_filter": ["github.com", "gitlab.com"]} + ) + assert data["query"] == "latest AI developments (site:github.com OR site:gitlab.com)" + + +def test_transform_search_request_forwards_scavio_params() -> None: + data: Final = _config().transform_search_request( + "q", {"hl": "fr", "location": "Paris,France", "device": "mobile", "start": 10, "time_period": "last_week"} + ) + assert (data["hl"], data["location"], data["device"], data["start"], data["time_period"]) == ( + "fr", + "Paris,France", + "mobile", + 10, + "last_week", + ) + + +def test_transform_search_response_maps_organic_results_in_order() -> None: + resp: Final = _config().transform_search_response( + _resp({"organic_results": [_result(title=t, link=f"https://{t}.com") for t in ("first", "second", "third")]}), + logging_obj=_logging_obj(), + ) + assert [(r.title, r.url) for r in resp.results] == [ + ("first", "https://first.com"), + ("second", "https://second.com"), + ("third", "https://third.com"), + ] + assert resp.results[0].snippet == "Test snippet" + assert resp.results[0].date is None + + +def test_transform_search_response_result_without_snippet_does_not_fail_the_call() -> None: + no_snippet: Final = {"position": 4, "title": "Forum thread", "link": "https://forum.example"} + resp: Final = _config().transform_search_response( + _resp({"organic_results": [no_snippet, _result()]}), logging_obj=_logging_obj() + ) + assert [r.snippet for r in resp.results] == ["", "Test snippet"] + + +@pytest.mark.parametrize("max_results, expected_count", [(2, 2), (10, 5), (0, 5), (True, 5), (None, 5)]) +def test_transform_search_response_caps_to_max_results(max_results: int | bool | None, expected_count: int) -> None: + resp: Final = _config().transform_search_response( + _resp({"organic_results": [_result(title=str(i)) for i in range(5)]}), + logging_obj=_logging_obj(), + optional_params={"max_results": max_results}, + ) + assert [r.title for r in resp.results] == [str(i) for i in range(expected_count)] + + +def test_transform_search_response_zero_hits() -> None: + payload: Final[Mapping[str, object]] = { + "search_information": {"total_results": 0}, + "organic_results": [], + "credits_used": 1, + } + assert _config().transform_search_response(_resp(payload), logging_obj=_logging_obj()).results == [] + + +@pytest.mark.parametrize( + "body", + [ + "502 Bad Gateway", + '{"organic_results": ["garbage"]}', + '{"organic_results": {"unexpected": "shape"}}', + '{"organic_results": null}', + "{}", + ], +) +def test_transform_search_response_malformed_body_raises_instead_of_reporting_empty(body: str) -> None: + with pytest.raises(Exception, match="Scavio Search"): + _config().transform_search_response(_resp(body, status_code=502), logging_obj=_logging_obj()) + + +def test_get_error_class_unwraps_scavio_error_envelope() -> None: + error: Final = _config().get_error_class(error_message='{"error":"Invalid API key"}', status_code=401, headers={}) + assert getattr(error, "status_code", None) == 401 + assert str(error) == "Scavio Search: Invalid API key. See https://scavio.dev/docs/search-api for details." + + +@pytest.mark.parametrize("body", ["502 Bad Gateway", '{"error": null}']) +def test_get_error_class_falls_back_to_the_raw_body(body: str) -> None: + assert f"Scavio Search: {body}." in str(_config().get_error_class(body, status_code=502, headers={})) + + +def test_search_handler_sends_mapped_request_and_parses_response() -> None: + stub: Final = _ScavioStub(200, LIVE_RESPONSE) + client: Final = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(stub))) + + response: Final = BaseLLMHTTPHandler().search( + query="open source llm gateway", + optional_params={"max_results": 1, "country": "US", "search_domain_filter": ["github.com"]}, + timeout=10.0, + logging_obj=_logging_obj(), + api_key="test-api-key", + api_base=None, + custom_llm_provider="scavio", + client=client, + provider_config=_config(), + ) + + assert isinstance(response, SearchResponse) + assert [str(r.url) for r in stub.requests] == [SCAVIO_URL] + assert stub.requests[0].headers["Authorization"] == "Bearer test-api-key" + assert json.loads(stub.requests[0].content) == { + "resolve_ai_overview": False, + "gl": "us", + "query": "open source llm gateway site:github.com", + } + assert [(r.title, r.url, r.snippet) for r in response.results] == [ + ( + "LiteLLM - GitHub", + "https://github.com/BerriAI/litellm", + "Python SDK, Proxy Server (AI Gateway) to call 100+ LLM APIs in OpenAI format.", + ) + ] + + +def test_search_handler_surfaces_scavio_error_message() -> None: + stub: Final = _ScavioStub(401, {"error": "Invalid API key"}) + client: Final = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(stub))) + + with pytest.raises(Exception, match="Scavio Search: Invalid API key"): + BaseLLMHTTPHandler().search( + query="q", + optional_params={}, + timeout=10.0, + logging_obj=_logging_obj(), + api_key="wrong-key", + api_base=None, + custom_llm_provider="scavio", + client=client, + provider_config=_config(), + ) + assert len(stub.requests) == 1 + + +@pytest.mark.asyncio +async def test_async_search_handler_sends_mapped_request_and_parses_response() -> None: + stub: Final = _ScavioStub(200, LIVE_RESPONSE) + client: Final = AsyncHTTPHandler(transport=httpx.MockTransport(stub)) + + response: Final = await BaseLLMHTTPHandler().async_search( + query=["open source", "llm gateway"], + optional_params={"max_results": 1}, + timeout=10.0, + logging_obj=_logging_obj(), + api_key="test-api-key", + api_base=None, + custom_llm_provider="scavio", + client=client, + provider_config=_config(), + ) + + assert json.loads(stub.requests[0].content)["query"] == "open source llm gateway" + assert [r.title for r in response.results] == ["LiteLLM - GitHub"] + + +def test_search_cost_comes_from_the_cost_map() -> None: + expected: Final[float] = litellm.model_cost["scavio/search"]["input_cost_per_query"] + assert expected > 0 + assert search_provider_cost_per_query(model="scavio/search", custom_llm_provider="scavio") == (expected, 0.0)