jpmschweitzerandClaude Opus 4.5
61863ff597
Build and Push / build (release) Successful in 1m2s
- Add /rag/search endpoint for web, news, and image search via SearXNG - Add /content/extract and /content/extract/batch endpoints - Add ContentExtractor client using Trafilatura for content extraction - Enhance HybridRAG web search with full content extraction - Add Redis caching for search results - Add new configuration options for search and extraction timeouts 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
370 lines
12 KiB
Python
370 lines
12 KiB
Python
"""Tests for RAG search service and endpoints."""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from src.models.rag_search import (
|
|
SearchType,
|
|
RAGSearchRequest,
|
|
RAGSearchResult,
|
|
RAGSearchResponse,
|
|
)
|
|
from src.services.rag_search_service import RAGSearchService, extract_domain
|
|
|
|
|
|
class TestExtractDomain:
|
|
"""Tests for domain extraction utility."""
|
|
|
|
def test_extract_simple_domain(self):
|
|
"""Test extracting domain from simple URL."""
|
|
assert extract_domain("https://example.com/page") == "example.com"
|
|
|
|
def test_extract_domain_with_www(self):
|
|
"""Test extracting domain removes www prefix."""
|
|
assert extract_domain("https://www.example.com/page") == "example.com"
|
|
|
|
def test_extract_domain_with_subdomain(self):
|
|
"""Test extracting domain preserves subdomains."""
|
|
assert extract_domain("https://blog.example.com/post") == "blog.example.com"
|
|
|
|
def test_extract_domain_invalid_url(self):
|
|
"""Test extracting domain from invalid URL returns empty string."""
|
|
# urlparse returns empty netloc for invalid URLs
|
|
assert extract_domain("not-a-url") == ""
|
|
|
|
|
|
class TestRAGSearchModels:
|
|
"""Tests for RAG search Pydantic models."""
|
|
|
|
def test_search_request_defaults(self):
|
|
"""Test RAGSearchRequest with default values."""
|
|
request = RAGSearchRequest(query="test query")
|
|
|
|
assert request.query == "test query"
|
|
assert request.search_type == SearchType.WEB
|
|
assert request.limit == 10
|
|
|
|
def test_search_request_custom_values(self):
|
|
"""Test RAGSearchRequest with custom values."""
|
|
request = RAGSearchRequest(
|
|
query="news about AI",
|
|
search_type=SearchType.NEWS,
|
|
limit=5,
|
|
user="custom_user"
|
|
)
|
|
|
|
assert request.query == "news about AI"
|
|
assert request.search_type == SearchType.NEWS
|
|
assert request.limit == 5
|
|
assert request.user == "custom_user"
|
|
|
|
def test_search_result(self):
|
|
"""Test RAGSearchResult model."""
|
|
result = RAGSearchResult(
|
|
title="Test Article",
|
|
url="https://example.com/article",
|
|
content="Full article content",
|
|
snippet="Article snippet...",
|
|
source="example.com",
|
|
published_date="2024-01-15"
|
|
)
|
|
|
|
assert result.title == "Test Article"
|
|
assert result.source == "example.com"
|
|
assert result.published_date == "2024-01-15"
|
|
|
|
def test_search_response(self):
|
|
"""Test RAGSearchResponse model."""
|
|
response = RAGSearchResponse(
|
|
query="test",
|
|
search_type=SearchType.WEB,
|
|
results=[],
|
|
total_results=0,
|
|
search_time_ms=100,
|
|
sources_summary=""
|
|
)
|
|
|
|
assert response.query == "test"
|
|
assert response.total_results == 0
|
|
assert response.search_time_ms == 100
|
|
|
|
|
|
class TestRAGSearchService:
|
|
"""Tests for RAGSearchService."""
|
|
|
|
@pytest.fixture
|
|
def mock_searxng_client(self):
|
|
"""Create mock SearXNG client."""
|
|
client = MagicMock()
|
|
client.search_general = AsyncMock(return_value=[
|
|
{
|
|
"title": "Test Result 1",
|
|
"url": "https://example.com/1",
|
|
"content": "Snippet 1",
|
|
"publishedDate": "2024-01-15"
|
|
},
|
|
{
|
|
"title": "Test Result 2",
|
|
"url": "https://example.com/2",
|
|
"content": "Snippet 2",
|
|
"publishedDate": None
|
|
}
|
|
])
|
|
client.search_news = AsyncMock(return_value=[])
|
|
client.search_images = AsyncMock(return_value=[])
|
|
return client
|
|
|
|
@pytest.fixture
|
|
def mock_content_extractor(self):
|
|
"""Create mock ContentExtractor."""
|
|
from src.models.content import ContentExtractionResult
|
|
|
|
extractor = MagicMock()
|
|
extractor.extract_batch = AsyncMock(return_value=[
|
|
ContentExtractionResult(
|
|
url="https://example.com/1",
|
|
content="Full extracted content 1",
|
|
success=True
|
|
),
|
|
ContentExtractionResult(
|
|
url="https://example.com/2",
|
|
content="Full extracted content 2",
|
|
success=True
|
|
)
|
|
])
|
|
return extractor
|
|
|
|
@pytest.fixture
|
|
def mock_redis_client(self):
|
|
"""Create mock Redis client."""
|
|
redis = MagicMock()
|
|
redis.get = AsyncMock(return_value=None) # No cache hit
|
|
redis.setex = AsyncMock()
|
|
return redis
|
|
|
|
@pytest.fixture
|
|
def mock_settings(self):
|
|
"""Create mock settings."""
|
|
settings = MagicMock()
|
|
settings.search_cache_ttl = 300
|
|
settings.search_default_limit = 10
|
|
return settings
|
|
|
|
@pytest.fixture
|
|
def rag_search_service(
|
|
self,
|
|
mock_searxng_client,
|
|
mock_content_extractor,
|
|
mock_redis_client,
|
|
mock_settings
|
|
):
|
|
"""Create RAGSearchService with mocked dependencies."""
|
|
return RAGSearchService(
|
|
searxng_client=mock_searxng_client,
|
|
content_extractor=mock_content_extractor,
|
|
redis_client=mock_redis_client,
|
|
settings=mock_settings
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_basic(self, rag_search_service, mock_searxng_client):
|
|
"""Test basic web search."""
|
|
response = await rag_search_service.search(
|
|
query="test query",
|
|
search_type=SearchType.WEB,
|
|
limit=10
|
|
)
|
|
|
|
assert response.query == "test query"
|
|
assert response.search_type == SearchType.WEB
|
|
assert len(response.results) == 2
|
|
assert response.total_results == 2
|
|
assert response.search_time_ms >= 0
|
|
|
|
mock_searxng_client.search_general.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_news(self, rag_search_service, mock_searxng_client):
|
|
"""Test news search type."""
|
|
mock_searxng_client.search_news.return_value = [
|
|
{"title": "News", "url": "https://news.com/1", "content": "News content"}
|
|
]
|
|
|
|
response = await rag_search_service.search(
|
|
query="latest news",
|
|
search_type=SearchType.NEWS
|
|
)
|
|
|
|
assert response.search_type == SearchType.NEWS
|
|
mock_searxng_client.search_news.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_images(self, rag_search_service, mock_searxng_client):
|
|
"""Test image search type."""
|
|
mock_searxng_client.search_images.return_value = [
|
|
{"title": "Image", "url": "https://images.com/1.jpg", "content": ""}
|
|
]
|
|
|
|
response = await rag_search_service.search(
|
|
query="cat photos",
|
|
search_type=SearchType.IMAGES
|
|
)
|
|
|
|
assert response.search_type == SearchType.IMAGES
|
|
mock_searxng_client.search_images.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_empty_query(self, rag_search_service):
|
|
"""Test search with empty query raises ValueError."""
|
|
with pytest.raises(ValueError, match="Query cannot be empty"):
|
|
await rag_search_service.search(query="", search_type=SearchType.WEB)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_caching_miss(
|
|
self,
|
|
rag_search_service,
|
|
mock_redis_client,
|
|
mock_searxng_client
|
|
):
|
|
"""Test search caches results on cache miss."""
|
|
mock_redis_client.get.return_value = None # Cache miss
|
|
|
|
await rag_search_service.search(query="test", search_type=SearchType.WEB)
|
|
|
|
# Should call SearXNG (cache miss)
|
|
mock_searxng_client.search_general.assert_called_once()
|
|
# Should cache result
|
|
mock_redis_client.setex.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_caching_hit(
|
|
self,
|
|
rag_search_service,
|
|
mock_redis_client,
|
|
mock_searxng_client
|
|
):
|
|
"""Test search returns cached results on cache hit."""
|
|
# Simulate cache hit
|
|
cached_response = RAGSearchResponse(
|
|
query="test",
|
|
search_type=SearchType.WEB,
|
|
results=[],
|
|
total_results=0,
|
|
search_time_ms=50,
|
|
sources_summary=""
|
|
)
|
|
mock_redis_client.get.return_value = cached_response.model_dump_json()
|
|
|
|
response = await rag_search_service.search(query="test", search_type=SearchType.WEB)
|
|
|
|
# Should NOT call SearXNG (cache hit)
|
|
mock_searxng_client.search_general.assert_not_called()
|
|
assert response.query == "test"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_content_extraction(
|
|
self,
|
|
rag_search_service,
|
|
mock_content_extractor
|
|
):
|
|
"""Test search extracts content from result URLs."""
|
|
response = await rag_search_service.search(
|
|
query="test",
|
|
search_type=SearchType.WEB
|
|
)
|
|
|
|
# Should have called content extractor
|
|
mock_content_extractor.extract_batch.assert_called_once()
|
|
|
|
# Results should have extracted content
|
|
for result in response.results:
|
|
assert result.content # Content should be populated
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_sources_summary(self, rag_search_service):
|
|
"""Test search generates sources summary."""
|
|
response = await rag_search_service.search(
|
|
query="test",
|
|
search_type=SearchType.WEB
|
|
)
|
|
|
|
assert response.sources_summary
|
|
assert "## Sources" in response.sources_summary
|
|
assert "[Test Result 1]" in response.sources_summary
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_search_limit(self, rag_search_service, mock_searxng_client):
|
|
"""Test search respects limit parameter."""
|
|
await rag_search_service.search(
|
|
query="test",
|
|
search_type=SearchType.WEB,
|
|
limit=5
|
|
)
|
|
|
|
# Check limit was passed to SearXNG
|
|
mock_searxng_client.search_general.assert_called_once_with(
|
|
query="test",
|
|
limit=5
|
|
)
|
|
|
|
|
|
class TestRAGSearchServiceIntegration:
|
|
"""Integration-style tests (still mocked but test more of the flow)."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_full_search_flow(self):
|
|
"""Test full search flow with all components mocked."""
|
|
from src.models.content import ContentExtractionResult
|
|
|
|
# Setup mocks
|
|
mock_searxng = MagicMock()
|
|
mock_searxng.search_general = AsyncMock(return_value=[
|
|
{
|
|
"title": "Python Tutorial",
|
|
"url": "https://python.org/tutorial",
|
|
"content": "Learn Python programming",
|
|
"publishedDate": "2024-01-10"
|
|
}
|
|
])
|
|
|
|
mock_extractor = MagicMock()
|
|
mock_extractor.extract_batch = AsyncMock(return_value=[
|
|
ContentExtractionResult(
|
|
url="https://python.org/tutorial",
|
|
title="Python Tutorial",
|
|
content="This is a comprehensive Python tutorial covering basics to advanced topics.",
|
|
success=True
|
|
)
|
|
])
|
|
|
|
mock_redis = MagicMock()
|
|
mock_redis.get = AsyncMock(return_value=None)
|
|
mock_redis.setex = AsyncMock()
|
|
|
|
mock_settings = MagicMock()
|
|
mock_settings.search_cache_ttl = 300
|
|
mock_settings.search_default_limit = 10
|
|
|
|
# Create service and execute search
|
|
service = RAGSearchService(
|
|
searxng_client=mock_searxng,
|
|
content_extractor=mock_extractor,
|
|
redis_client=mock_redis,
|
|
settings=mock_settings
|
|
)
|
|
|
|
response = await service.search(
|
|
query="python tutorial",
|
|
search_type=SearchType.WEB,
|
|
limit=10,
|
|
user="test_user"
|
|
)
|
|
|
|
# Verify response
|
|
assert response.query == "python tutorial"
|
|
assert len(response.results) == 1
|
|
assert response.results[0].title == "Python Tutorial"
|
|
assert response.results[0].source == "python.org"
|
|
assert "comprehensive Python tutorial" in response.results[0].content
|
|
assert response.results[0].snippet == "Learn Python programming"
|