from __future__ import annotations
import httpx
import pytest
from opensquilla.gateway.config import GatewayConfig
from opensquilla.search.providers.brave import BraveSearchProvider
from opensquilla.search.providers.duckduckgo import DuckDuckGoProvider
from opensquilla.search.providers.exa import ExaSearchProvider
from opensquilla.search.types import SearchProviderError, SearchResult
from opensquilla.tools.builtin import web
def test_gateway_config_accepts_search_api_key() -> None:
config = GatewayConfig(search_api_key="brave-test-key")
assert config.search_api_key == "brave-test-key"
def test_brave_provider_prefers_explicit_api_key(monkeypatch) -> None:
monkeypatch.delenv("BRAVE_SEARCH_API_KEY", raising=False)
provider = BraveSearchProvider(api_key="brave-test-key")
assert provider._api_key == "brave-test-key"
def test_brave_provider_strips_trailing_paste_punctuation(monkeypatch) -> None:
monkeypatch.delenv("BRAVE_SEARCH_API_KEY", raising=False)
provider = BraveSearchProvider(api_key="brave-test-key、")
assert provider._api_key == "brave-test-key"
def test_web_search_kwargs_pass_brave_api_key() -> None:
web.configure_search("brave", api_key="brave-test-key")
assert web._search_provider_kwargs("brave")["api_key"] == "brave-test-key"
def test_web_search_kwargs_pass_tavily_api_key() -> None:
web.configure_search("tavily", api_key="tavily-test-key")
assert web._search_provider_kwargs("tavily")["api_key"] == "tavily-test-key"
def test_web_search_kwargs_pass_exa_api_key() -> None:
web.configure_search("exa", api_key="exa-test-key")
assert web._search_provider_kwargs("exa")["api_key"] == "exa-test-key"
def test_exa_provider_prefers_explicit_api_key(monkeypatch) -> None:
monkeypatch.delenv("EXA_API_KEY", raising=False)
provider = ExaSearchProvider(api_key="exa-test-key")
assert provider._api_key == "exa-test-key"
@pytest.mark.asyncio
async def test_brave_provider_maps_provider_source_and_published_at() -> None:
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(
200,
json={
"web": {
"results": [
{
"title": "Brave title",
"url": "https://example.com/brave",
"description": "Brave snippet",
"age": "2026-06-19",
}
]
}
},
)
provider = BraveSearchProvider(
api_key="brave-test-key",
transport=httpx.MockTransport(handler),
)
result = (await provider.search("brave"))[0]
assert result.provider == "brave"
assert result.source == "brave"
assert result.published_at == "2026-06-19"
@pytest.mark.asyncio
async def test_brave_provider_passes_recency_as_freshness() -> None:
requests: list[httpx.Request] = []
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
return httpx.Response(200, json={"web": {"results": []}})
provider = BraveSearchProvider(
api_key="brave-test-key",
transport=httpx.MockTransport(handler),
)
await provider.search("brave", recency="week")
assert requests[0].url.params["freshness"] == "pw"
@pytest.mark.asyncio
async def test_duckduckgo_provider_maps_provider_and_source() -> None:
html = """
"""
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, text=html)
provider = DuckDuckGoProvider(transport=httpx.MockTransport(handler))
result = (await provider.search("duck"))[0]
assert result.provider == "duckduckgo"
assert result.source == "duckduckgo"
@pytest.mark.asyncio
async def test_duckduckgo_provider_treats_anomaly_challenge_as_blocked() -> None:
html = """
"""
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(202, text=html)
provider = DuckDuckGoProvider(transport=httpx.MockTransport(handler))
with pytest.raises(SearchProviderError) as exc_info:
await provider.search("duck")
assert exc_info.value.provider == "duckduckgo"
assert exc_info.value.kind == "blocked"
assert exc_info.value.retryable is True
@pytest.mark.asyncio
async def test_duckduckgo_provider_allows_real_no_results() -> None:
html = """
No results found for improbable query.
"""
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, text=html)
provider = DuckDuckGoProvider(transport=httpx.MockTransport(handler))
assert await provider.search("improbable query") == []
@pytest.mark.asyncio
async def test_duckduckgo_provider_treats_unparseable_empty_page_as_parse_error() -> None:
html = "DuckDuckGo"
def handler(request: httpx.Request) -> httpx.Response:
return httpx.Response(200, text=html)
provider = DuckDuckGoProvider(transport=httpx.MockTransport(handler))
with pytest.raises(SearchProviderError) as exc_info:
await provider.search("duck")
assert exc_info.value.provider == "duckduckgo"
assert exc_info.value.kind == "parse"
assert exc_info.value.retryable is True
@pytest.mark.asyncio
async def test_duckduckgo_provider_surfaces_network_failures() -> None:
def handler(request: httpx.Request) -> httpx.Response:
raise httpx.ConnectError("network down", request=request)
provider = DuckDuckGoProvider(transport=httpx.MockTransport(handler))
with pytest.raises(SearchProviderError) as exc_info:
await provider.search("duck")
assert exc_info.value.provider == "duckduckgo"
assert exc_info.value.kind == "network"
assert exc_info.value.retryable is True
def test_search_payload_keeps_lightweight_metadata() -> None:
payload = web._search_payload(
"python",
"tavily",
[
SearchResult(
title="Title",
url="https://Docs.Python.org/3/?utm_source=x#intro",
snippet="Snippet",
provider="tavily",
source="tavily",
published_at="2026-06-19",
score=0.9,
content="Full content must stay out",
raw_metadata={"debug": "must stay out"},
)
],
)
result = payload["results"][0]
assert result == {
"title": "Title",
"url": "https://Docs.Python.org/3/?utm_source=x#intro",
"snippet": "Snippet",
"provider": "tavily",
"published_at": "2026-06-19",
"score": 0.9,
"domain": "docs.python.org",
"canonical_url": "https://docs.python.org/3",
}