jpmschweitzerandClaude Fable 5
6ac1d15591
The homelab is retiring *.schweitz.internal and will rebind host ports to 127.0.0.1, so container-to-container traffic must use container names on the docker-dataplane network. Switch defaults from host IP:port to http://tatlock:8000 and http://searxng:8080 (SearXNG's internal port is 8080; 8087 is only the host-published port). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
218 lines
8.4 KiB
Python
218 lines
8.4 KiB
Python
"""
|
|
Tests for WebSearchTool.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
|
|
from src.domains.tools.search.web import WebSearchTool
|
|
|
|
|
|
class TestWebSearchTool:
|
|
"""Tests for WebSearchTool."""
|
|
|
|
@pytest.fixture
|
|
def tool(self):
|
|
return WebSearchTool(searxng_url="http://searxng:8080", timeout=5)
|
|
|
|
@pytest.fixture
|
|
def mock_search_response(self):
|
|
"""Sample SearXNG response."""
|
|
return {
|
|
"query": "test query",
|
|
"number_of_results": 3,
|
|
"results": [
|
|
{
|
|
"title": "First Result",
|
|
"url": "https://example.com/1",
|
|
"content": "This is the first result content.",
|
|
"engine": "google",
|
|
"publishedDate": "2024-01-15",
|
|
},
|
|
{
|
|
"title": "Second Result",
|
|
"url": "https://example.com/2",
|
|
"content": "This is the second result content.",
|
|
"engine": "brave",
|
|
"publishedDate": None,
|
|
},
|
|
{
|
|
"title": "Third Result",
|
|
"url": "https://example.com/3",
|
|
"content": "This is the third result content.",
|
|
"engine": "duckduckgo",
|
|
"publishedDate": "2024-01-10",
|
|
},
|
|
],
|
|
}
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_success(self, tool, mock_search_response):
|
|
"""Test successful search."""
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = mock_search_response
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.return_value = mock_response
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="test query", num_results=3)
|
|
|
|
assert result.success
|
|
assert "First Result" in result.data
|
|
assert "https://example.com/1" in result.data
|
|
assert result.metadata["result_count"] == 3
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_empty_query(self, tool):
|
|
"""Test empty query is rejected."""
|
|
result = await tool.execute(query="")
|
|
assert not result.success
|
|
assert "empty" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_whitespace_query(self, tool):
|
|
"""Test whitespace-only query is rejected."""
|
|
result = await tool.execute(query=" ")
|
|
assert not result.success
|
|
assert "empty" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_no_results(self, tool):
|
|
"""Test when search returns no results."""
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"query": "obscure", "results": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.return_value = mock_response
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="obscure nonexistent thing")
|
|
|
|
assert result.success
|
|
assert "No results" in result.data
|
|
assert result.metadata["result_count"] == 0
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_num_results_limit(self, tool, mock_search_response):
|
|
"""Test num_results limits output."""
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = mock_search_response
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.return_value = mock_response
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="test", num_results=2)
|
|
|
|
assert result.success
|
|
assert result.metadata["result_count"] == 2
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_num_results_clamped(self, tool, mock_search_response):
|
|
"""Test num_results is clamped to max."""
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = mock_search_response
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.return_value = mock_response
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
# Request 100 but max is 10
|
|
result = await tool.execute(query="test", num_results=100)
|
|
|
|
assert result.success
|
|
# Should only get 3 (what's in mock response, capped at 10)
|
|
assert result.metadata["result_count"] <= 10
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_timeout_error(self, tool):
|
|
"""Test timeout handling."""
|
|
import httpx
|
|
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.side_effect = httpx.TimeoutException("timeout")
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="test")
|
|
|
|
assert not result.success
|
|
assert "timed out" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_request_error(self, tool):
|
|
"""Test network error handling."""
|
|
import httpx
|
|
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.side_effect = httpx.RequestError("connection failed")
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="test")
|
|
|
|
assert not result.success
|
|
assert "failed" in result.error.lower()
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_includes_engines_metadata(self, tool, mock_search_response):
|
|
"""Test engines used are included in metadata."""
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = mock_search_response
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.return_value = mock_response
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="test", num_results=3)
|
|
|
|
assert result.success
|
|
engines = result.metadata.get("engines_used", [])
|
|
assert "google" in engines or "brave" in engines
|
|
|
|
@pytest.mark.anyio
|
|
async def test_search_with_categories(self, tool, mock_search_response):
|
|
"""Test category parameter is passed."""
|
|
with patch("httpx.AsyncClient") as mock_client:
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = mock_search_response
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
mock_instance = AsyncMock()
|
|
mock_instance.get.return_value = mock_response
|
|
mock_instance.__aenter__.return_value = mock_instance
|
|
mock_instance.__aexit__.return_value = None
|
|
mock_client.return_value = mock_instance
|
|
|
|
result = await tool.execute(query="test", categories="it")
|
|
|
|
assert result.success
|
|
# Verify get was called with categories parameter
|
|
call_args = mock_instance.get.call_args
|
|
assert "categories" in call_args.kwargs["params"]
|
|
assert call_args.kwargs["params"]["categories"] == "it"
|