"""Tests for Tavily web backend integration. Coverage: _tavily_request() — keyed Bearer vs keyless header, attribution, error bodies. _normalize_tavily_search_results() — search response normalization. _normalize_tavily_documents() — extract response normalization, failed_results. web_search_tool / web_extract_tool — Tavily dispatch paths. auto-detect ranking — keyed paid-band; keyless only when Tavily is selected. """ import json import os import asyncio import pytest from unittest.mock import patch, MagicMock from tests.tools.conftest import register_all_web_providers def _ok_response(payload=None): mock_response = MagicMock() mock_response.status_code = 200 mock_response.json.return_value = payload if payload is not None else {"results": []} mock_response.text = json.dumps(mock_response.json.return_value) return mock_response # ─── _tavily_request ───────────────────────────────────────────────────────── class TestTavilyRequest: """Test suite for the _tavily_request helper.""" def test_keyless_when_no_api_key(self): """No TAVILY_API_KEY → keyless header, no Authorization, no body key.""" mock_response = _ok_response() with patch.dict(os.environ, {}, clear=False): os.environ.pop("TAVILY_API_KEY", None) with patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response) as mock_post: from plugins.web.tavily.provider import _tavily_request _tavily_request("search", {"query": "test"}) mock_post.assert_called_once() headers = mock_post.call_args.kwargs["headers"] payload = mock_post.call_args.kwargs["json"] assert headers["X-Client-Name"] == "hermes-agent" assert headers["X-Tavily-Access-Mode"] == "keyless" assert "Authorization" not in headers assert "api_key" not in payload assert payload["query"] == "test" assert "api.tavily.com/search" in mock_post.call_args.args[0] def test_keyed_uses_bearer_not_body(self): """TAVILY_API_KEY → Bearer auth, attribution, no body api_key.""" mock_response = _ok_response() with patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test-key"}): with patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response) as mock_post: from plugins.web.tavily.provider import _tavily_request _tavily_request("search", {"query": "hello"}) mock_post.assert_called_once() headers = mock_post.call_args.kwargs["headers"] payload = mock_post.call_args.kwargs["json"] assert headers == { "X-Client-Name": "hermes-agent", "Authorization": "Bearer tvly-test-key", } assert "X-Tavily-Access-Mode" not in headers assert "api_key" not in payload assert payload["query"] == "hello" assert "api.tavily.com/search" in mock_post.call_args.args[0] def test_http_error_surfaces_response_body(self): """Non-2xx responses raise ValueError with Tavily's response body.""" mock_response = MagicMock() mock_response.status_code = 429 mock_response.text = "Rate limit hit. Sign up for a free API key at https://app.tavily.com" mock_response.json.return_value = {} with patch.dict(os.environ, {}, clear=False): os.environ.pop("TAVILY_API_KEY", None) with patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response): from plugins.web.tavily.provider import _tavily_request with pytest.raises(ValueError, match="Rate limit hit"): _tavily_request("search", {"query": "test"}) # ─── _normalize_tavily_search_results ───────────────────────────────────────── class TestNormalizeTavilySearchResults: """Test search result normalization.""" def test_basic_normalization(self): from plugins.web.tavily.provider import _normalize_tavily_search_results raw = { "results": [ {"title": "Python Docs", "url": "https://docs.python.org", "content": "Official docs", "score": 0.9}, {"title": "Tutorial", "url": "https://example.com", "content": "A tutorial", "score": 0.8}, ] } result = _normalize_tavily_search_results(raw) assert result["success"] is True web = result["data"]["web"] assert len(web) == 2 assert web[0]["title"] == "Python Docs" assert web[0]["url"] == "https://docs.python.org" assert web[0]["description"] == "Official docs" assert web[0]["position"] == 1 assert web[1]["position"] == 2 def test_missing_fields(self): from plugins.web.tavily.provider import _normalize_tavily_search_results result = _normalize_tavily_search_results({"results": [{}]}) web = result["data"]["web"] assert web[0]["title"] == "" assert web[0]["url"] == "" assert web[0]["description"] == "" # ─── _normalize_tavily_documents ────────────────────────────────────────────── class TestNormalizeTavilyDocuments: """Test extract document normalization.""" def test_basic_document(self): from plugins.web.tavily.provider import _normalize_tavily_documents raw = { "results": [{ "url": "https://example.com", "title": "Example", "raw_content": "Full page content here", }] } docs = _normalize_tavily_documents(raw) assert len(docs) == 1 assert docs[0]["url"] == "https://example.com" assert docs[0]["title"] == "Example" assert docs[0]["content"] == "Full page content here" assert docs[0]["raw_content"] == "Full page content here" assert docs[0]["metadata"]["sourceURL"] == "https://example.com" def test_missing_result_url_is_not_attributed_to_requested_url(self): from plugins.web.tavily.provider import _normalize_tavily_documents raw = {"results": [{"content": "data"}]} docs = _normalize_tavily_documents(raw) assert docs[0]["url"] == "" class TestWebExtractCacheAttribution: """Only cache content under the URL reported by the extract provider.""" def test_partial_result_caches_under_its_own_requested_url(self): from tools import web_tools_extract as wte class _PartialProvider: name = "tavily" async def extract(self, urls, format=None): return [ {"url": "https://unrequested.example/page", "raw_content": "foreign page"}, {"url": urls[1], "raw_content": "second page", "title": "Second"}, ] urls = ["https://example.com/first", "https://example.com/second"] with patch("tools.web_result_cache.extract_cache_put") as cache_put: asyncio.run(wte._dispatch_extract(_PartialProvider(), urls, None)) cache_put.assert_called_once_with( "https://example.com/second", "second page", "Second", format=None, provider="tavily" ) def test_redirected_page_caches_under_requested_source_url(self): """Keenable/Firecrawl report the post-redirect address in ``url`` and the requested URL in ``metadata.sourceURL``; the cache key must stay the requested URL, never the redirect target. The first requested page is omitted, so positional pairing would file the redirected page under the wrong requested URL.""" from tools import web_tools_extract as wte class _RedirectProvider: name = "keenable" async def extract(self, urls, format=None): return [{"url": "https://www.example.com/moved", "raw_content": "moved page", "title": "Moved", "metadata": {"sourceURL": urls[1]}}] urls = ["https://example.com/failed", "https://example.com/old"] with patch("tools.web_result_cache.extract_cache_put") as cache_put: asyncio.run(wte._dispatch_extract(_RedirectProvider(), urls, None)) cache_put.assert_called_once_with( "https://example.com/old", "moved page", "Moved", format=None, provider="keenable" ) # ─── availability / auto-detect ─────────────────────────────────────────────── class TestTavilyAvailability: """Keyed Tavily stays in the paid band; keyless only when selected.""" def test_is_available_without_key(self): from plugins.web.tavily.provider import TavilyWebSearchProvider with patch.dict(os.environ, {}, clear=False): os.environ.pop("TAVILY_API_KEY", None) assert TavilyWebSearchProvider().is_available() is False def test_is_backend_available_without_key(self): from tools.web_tools import _is_backend_available with patch("tools.web_tools._load_web_config", return_value={}), \ patch.dict(os.environ, {}, clear=False): os.environ.pop("TAVILY_API_KEY", None) assert _is_backend_available("tavily") is False def test_is_backend_available_when_configured_without_key(self): from tools.web_tools import _is_backend_available with patch("tools.web_tools._load_web_config", return_value={"backend": "tavily"}), \ patch.dict(os.environ, {}, clear=False): os.environ.pop("TAVILY_API_KEY", None) assert _is_backend_available("tavily") is True def test_keyless_does_not_preempt_managed_firecrawl(self): """No TAVILY_API_KEY + Nous gateway ready → firecrawl, not keyless tavily.""" from tools.web_tools import _get_backend with patch("tools.web_tools._load_web_config", return_value={}), \ patch("tools.web_tools._is_tool_gateway_ready", return_value=True), \ patch("tools.web_tools._ddgs_package_importable", return_value=False): os.environ.pop("TAVILY_API_KEY", None) assert _get_backend() == "firecrawl" def test_keyless_does_not_preempt_ddgs(self): from tools.web_tools import _get_backend with patch("tools.web_tools._load_web_config", return_value={}), \ patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \ patch("tools.web_tools._ddgs_package_importable", return_value=True): os.environ.pop("TAVILY_API_KEY", None) assert _get_backend() == "ddgs" def test_no_keys_defaults_to_firecrawl(self): """Keyless tier disabled: zero-credential resolve hits the legacy firecrawl sentinel. (With the tier on — the default — it resolves to the Exa/Parallel keyless split; see test_web_keyless_fallback.py.) """ from tools.web_tools import _get_backend with patch("tools.web_tools._load_web_config", return_value={}), \ patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \ patch("tools.web_tools._ddgs_package_importable", return_value=False), \ patch("tools.web_tools._list_registered_web_providers", return_value=[]), \ patch("agent.web_search_registry._keyless_tier_enabled", return_value=False): os.environ.pop("TAVILY_API_KEY", None) assert _get_backend() == "firecrawl" def test_explicit_search_backend_tavily_without_key(self): """web.search_backend=tavily sticks even with no TAVILY_API_KEY.""" from tools.web_tools import _get_search_backend with patch("tools.web_tools._load_web_config", return_value={"backend": "firecrawl", "search_backend": "tavily"}), \ patch("tools.web_tools._is_tool_gateway_ready", return_value=True): os.environ.pop("TAVILY_API_KEY", None) assert _get_search_backend() == "tavily" def test_check_web_api_key_when_tavily_configured_without_key(self): from tools.web_tools import check_web_api_key with patch("tools.web_tools._load_web_config", return_value={"backend": "tavily"}), \ patch("tools.web_tools._is_tool_gateway_ready", return_value=False), \ patch("tools.web_tools.check_firecrawl_api_key", return_value=False), \ patch("tools.web_tools._ddgs_package_importable", return_value=False), \ patch("agent.web_search_registry.get_active_search_provider", return_value=None), \ patch("agent.web_search_registry.get_active_extract_provider", return_value=None): os.environ.pop("TAVILY_API_KEY", None) assert check_web_api_key() is True # ─── web_search_tool (Tavily dispatch) ──────────────────────────────────────── class TestWebSearchTavily: """Test web_search_tool dispatch to Tavily.""" _register_providers = staticmethod(register_all_web_providers) @pytest.fixture(autouse=True) def _populate_web_registry(self): self._register_providers() yield from agent.web_search_registry import _reset_for_tests _reset_for_tests() def test_search_dispatches_to_tavily(self): mock_response = _ok_response({ "results": [{"title": "Result", "url": "https://r.com", "content": "desc", "score": 0.9}] }) with patch("tools.web_tools._get_backend", return_value="tavily"), \ patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}), \ patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response), \ patch("tools.interrupt.is_interrupted", return_value=False): from tools.web_tools import web_search_tool result = json.loads(web_search_tool("test query", limit=3)) assert result["success"] is True assert len(result["data"]["web"]) == 1 assert result["data"]["web"][0]["title"] == "Result" def test_search_keyless_dispatch(self): """Opt-in keyless Tavily hits Tavily's own endpoint, not the ring.""" mock_response = _ok_response({ "results": [{"title": "Result", "url": "https://r.com", "content": "desc"}] }) with patch("tools.web_tools._get_backend", return_value="tavily"), \ patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response) as mock_post, \ patch("tools.interrupt.is_interrupted", return_value=False): os.environ.pop("TAVILY_API_KEY", None) from tools.web_tools import web_search_tool result = json.loads(web_search_tool("test query")) assert result["success"] is True headers = mock_post.call_args.kwargs["headers"] assert headers["X-Tavily-Access-Mode"] == "keyless" assert headers["X-Client-Name"] == "hermes-agent" assert "Authorization" not in headers assert "api.tavily.com/search" in mock_post.call_args.args[0] # ─── web_extract_tool (Tavily dispatch) ─────────────────────────────────────── class TestWebExtractTavily: """Test web_extract_tool dispatch to Tavily.""" _register_providers = staticmethod(register_all_web_providers) @pytest.fixture(autouse=True) def _populate_web_registry(self): self._register_providers() yield from agent.web_search_registry import _reset_for_tests _reset_for_tests() def test_extract_dispatches_to_tavily(self): mock_response = _ok_response({ "results": [{"url": "https://example.com", "raw_content": "Extracted content", "title": "Page"}] }) async def _allow_ssrf(_url: str) -> bool: return True with patch("tools.web_tools._get_backend", return_value="tavily"), \ patch.dict(os.environ, {"TAVILY_API_KEY": "tvly-test"}), \ patch("plugins.web.tavily.provider.httpx.post", return_value=mock_response), \ patch("tools.web_tools.async_is_safe_url", _allow_ssrf): from tools.web_tools import web_extract_tool result = json.loads(asyncio.get_event_loop().run_until_complete( web_extract_tool(["https://example.com"]) )) assert "results" in result assert len(result["results"]) == 1 assert result["results"][0]["url"] == "https://example.com" assert "Extracted content" in result["results"][0]["content"]