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>
184 lines
6.6 KiB
Python
184 lines
6.6 KiB
Python
"""Tests for ContentExtractor client."""
|
|
|
|
import pytest
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from src.clients.content_extractor import ContentExtractor
|
|
from src.models.content import ContentExtractionResult
|
|
|
|
|
|
@pytest.fixture
|
|
def content_extractor():
|
|
"""Create ContentExtractor with test configuration."""
|
|
return ContentExtractor(timeout=5, max_length=2000)
|
|
|
|
|
|
class TestContentExtractor:
|
|
"""Tests for ContentExtractor client."""
|
|
|
|
def test_init(self, content_extractor):
|
|
"""Test ContentExtractor initialization."""
|
|
assert content_extractor.timeout == 5
|
|
assert content_extractor.max_length == 2000
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_success(self, content_extractor):
|
|
"""Test successful content extraction."""
|
|
test_url = "https://example.com/article"
|
|
test_content = "This is the extracted article content."
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.fetch_url.return_value = "<html><body>Test</body></html>"
|
|
mock_traf.extract.return_value = test_content
|
|
mock_traf.bare_extraction.return_value = {
|
|
"title": "Test Article",
|
|
"author": "John Doe",
|
|
"date": "2024-01-15",
|
|
"language": "en"
|
|
}
|
|
|
|
result = await content_extractor.extract(test_url)
|
|
|
|
assert result.success is True
|
|
assert result.url == test_url
|
|
assert result.content == test_content
|
|
assert result.error is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_fetch_failure(self, content_extractor):
|
|
"""Test extraction when URL fetch fails."""
|
|
test_url = "https://example.com/nonexistent"
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.fetch_url.return_value = None
|
|
|
|
result = await content_extractor.extract(test_url)
|
|
|
|
assert result.success is False
|
|
assert result.url == test_url
|
|
assert result.content == ""
|
|
assert "Failed to fetch URL" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_no_content(self, content_extractor):
|
|
"""Test extraction when page has no extractable content."""
|
|
test_url = "https://example.com/empty"
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.fetch_url.return_value = "<html><body></body></html>"
|
|
mock_traf.extract.return_value = None
|
|
|
|
result = await content_extractor.extract(test_url)
|
|
|
|
assert result.success is False
|
|
assert "No content extracted" in result.error
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_max_length_truncation(self, content_extractor):
|
|
"""Test that content is truncated to max length."""
|
|
test_url = "https://example.com/long-article"
|
|
# Content longer than max_length (2000)
|
|
long_content = "x" * 3000
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.fetch_url.return_value = "<html><body>Test</body></html>"
|
|
mock_traf.extract.return_value = long_content
|
|
mock_traf.bare_extraction.return_value = {}
|
|
|
|
result = await content_extractor.extract(test_url)
|
|
|
|
assert result.success is True
|
|
assert len(result.content) <= content_extractor.max_length + 3 # +3 for "..."
|
|
assert result.content.endswith("...")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_batch(self, content_extractor):
|
|
"""Test batch extraction of multiple URLs."""
|
|
test_urls = [
|
|
"https://example.com/article1",
|
|
"https://example.com/article2",
|
|
"https://example.com/article3"
|
|
]
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.fetch_url.return_value = "<html><body>Test</body></html>"
|
|
mock_traf.extract.return_value = "Extracted content"
|
|
mock_traf.bare_extraction.return_value = {}
|
|
|
|
results = await content_extractor.extract_batch(test_urls)
|
|
|
|
assert len(results) == 3
|
|
for i, result in enumerate(results):
|
|
assert result.url == test_urls[i]
|
|
assert result.success is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_timeout(self):
|
|
"""Test extraction timeout handling."""
|
|
import time
|
|
|
|
test_url = "https://example.com/slow"
|
|
|
|
# Create an extractor with very short timeout
|
|
fast_extractor = ContentExtractor(timeout=0.001, max_length=2000)
|
|
|
|
def slow_fetch(url):
|
|
time.sleep(1) # Sleep synchronously (this runs in thread pool)
|
|
return "<html></html>"
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.fetch_url = slow_fetch
|
|
|
|
result = await fast_extractor.extract(test_url)
|
|
|
|
assert result.success is False
|
|
assert "timed out" in result.error.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_extract_from_html(self, content_extractor):
|
|
"""Test extraction from raw HTML."""
|
|
test_html = "<html><body><article>Article content here.</article></body></html>"
|
|
|
|
with patch('src.clients.content_extractor.trafilatura') as mock_traf:
|
|
mock_traf.extract.return_value = "Article content here."
|
|
mock_traf.bare_extraction.return_value = {"title": "Test"}
|
|
|
|
result = await content_extractor.extract_from_html(test_html, url="https://example.com")
|
|
|
|
assert result.success is True
|
|
assert result.content == "Article content here."
|
|
|
|
|
|
class TestContentExtractionResult:
|
|
"""Tests for ContentExtractionResult model."""
|
|
|
|
def test_success_result(self):
|
|
"""Test creating a successful result."""
|
|
result = ContentExtractionResult(
|
|
url="https://example.com",
|
|
title="Test Article",
|
|
content="Article content",
|
|
author="John Doe",
|
|
date="2024-01-15",
|
|
language="en",
|
|
success=True,
|
|
error=None
|
|
)
|
|
|
|
assert result.url == "https://example.com"
|
|
assert result.success is True
|
|
assert result.error is None
|
|
|
|
def test_failure_result(self):
|
|
"""Test creating a failure result."""
|
|
result = ContentExtractionResult(
|
|
url="https://example.com/error",
|
|
content="",
|
|
success=False,
|
|
error="Failed to fetch URL"
|
|
)
|
|
|
|
assert result.url == "https://example.com/error"
|
|
assert result.success is False
|
|
assert result.error == "Failed to fetch URL"
|