test_client.pyā¢8.71 kB
"""Tests for the SearXNG client."""
import pytest
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
from searxng_mcp_server.client import SearXNGClient, SearchResponse, SearchResult
@pytest.fixture
def mock_response_data():
"""Mock response data from SearXNG API."""
return {
"query": "test query",
"number_of_results": 2,
"results": [
{
"title": "Test Result 1",
"url": "https://example.com/1",
"content": "Test content 1",
"engine": "google",
"category": "general",
"score": 0.9,
},
{
"title": "Test Result 2",
"url": "https://example.com/2",
"content": "Test content 2",
"engine": "bing",
"category": "general",
"score": 0.8,
},
],
"suggestions": ["test suggestion 1", "test suggestion 2"],
"answers": [],
"infoboxes": [],
"unresponsive_engines": [],
}
@pytest.fixture
def mock_config_data():
"""Mock config data from SearXNG API."""
return {
"autocomplete": "",
"categories": ["general", "images", "news"],
"default_locale": "",
"default_theme": "simple",
"engines": [
{
"categories": ["general"],
"enabled": True,
"name": "google",
"shortcut": "go",
}
],
"instance_name": "SearXNG",
"locales": {"en": "English"},
"plugins": [],
"safe_search": 0,
}
class TestSearXNGClient:
"""Test cases for SearXNGClient."""
@pytest.mark.asyncio
async def test_search_success(self, mock_response_data):
"""Test successful search request."""
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
# Setup mock response
mock_response = MagicMock()
mock_response.json.return_value = mock_response_data
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
# Execute search
result = await client.search("test query")
# Verify request was made correctly
mock_get.assert_called_once()
call_args = mock_get.call_args
assert "https://example.com/search" in str(call_args)
# Verify response
assert isinstance(result, SearchResponse)
assert result.query == "test query"
assert result.number_of_results == 2
assert len(result.results) == 2
assert result.results[0].title == "Test Result 1"
assert result.suggestions == ["test suggestion 1", "test suggestion 2"]
@pytest.mark.asyncio
async def test_search_with_filters(self, mock_response_data):
"""Test search with category and engine filters."""
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
mock_response = MagicMock()
mock_response.json.return_value = mock_response_data
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
await client.search(
query="test",
categories=["images", "news"],
engines=["google", "bing"],
language="en",
page=2,
time_range="month",
safesearch=1,
)
# Verify request parameters
call_args = mock_get.call_args
params = call_args.kwargs["params"]
assert params["q"] == "test"
assert params["categories"] == "images,news"
assert params["engines"] == "google,bing"
assert params["language"] == "en"
assert params["pageno"] == 2
assert params["time_range"] == "month"
assert params["safesearch"] == 1
@pytest.mark.asyncio
async def test_get_suggestions_success(self):
"""Test successful suggestions request."""
mock_suggestions = ["suggestion 1", "suggestion 2", "suggestion 3"]
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
mock_response = MagicMock()
mock_response.json.return_value = mock_suggestions
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = await client.get_suggestions("test")
mock_get.assert_called_once()
assert result == mock_suggestions
@pytest.mark.asyncio
async def test_health_check_success(self):
"""Test successful health check."""
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
mock_response = MagicMock()
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = await client.health_check()
assert result["status"] == "ok"
assert "healthy" in result["message"]
@pytest.mark.asyncio
async def test_health_check_failure(self):
"""Test health check with failure."""
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
mock_get.side_effect = httpx.HTTPError("Connection failed")
result = await client.health_check()
assert result["status"] == "error"
assert "failed" in result["message"].lower()
@pytest.mark.asyncio
async def test_get_config_success(self, mock_config_data):
"""Test successful config retrieval."""
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
mock_response = MagicMock()
mock_response.json.return_value = mock_config_data
mock_response.raise_for_status = MagicMock()
mock_get.return_value = mock_response
result = await client.get_config()
assert result["instance_name"] == "SearXNG"
assert "general" in result["categories"]
assert len(result["engines"]) > 0
@pytest.mark.asyncio
async def test_client_without_context_manager(self):
"""Test that client raises error when used without context manager."""
client = SearXNGClient("https://example.com")
with pytest.raises(RuntimeError, match="Client not initialized"):
_ = client.client
@pytest.mark.asyncio
async def test_search_http_error(self):
"""Test search handling HTTP errors."""
async with SearXNGClient("https://example.com") as client:
with patch.object(
client.client,
"get",
new_callable=AsyncMock,
) as mock_get:
mock_get.side_effect = httpx.HTTPStatusError(
"Not found",
request=MagicMock(),
response=MagicMock(status_code=404),
)
with pytest.raises(httpx.HTTPStatusError):
await client.search("test")
def test_client_initialization(self):
"""Test client initialization with various parameters."""
client = SearXNGClient(
base_url="https://example.com/",
timeout=60.0,
verify_ssl=False,
)
assert client.base_url == "https://example.com"
assert client.timeout == 60.0
assert client.verify_ssl is False