185 lines
7 KiB
Python
185 lines
7 KiB
Python
"""Unit tests for FetchClient with mocked httpx."""
|
|
|
|
import contextlib
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import pytest
|
|
|
|
from web.config import WebSettings
|
|
from web.fetch.client import FetchClient
|
|
from web.schemas.fetch import FetchRequest
|
|
|
|
|
|
@pytest.fixture
|
|
def settings() -> WebSettings:
|
|
"""Create test settings."""
|
|
return WebSettings(
|
|
searxng_base_url="http://localhost:55100",
|
|
llm_base_url="http://localhost:14011",
|
|
external_url="http://localhost:51100",
|
|
fetch_min_text_length=200,
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def client(settings: WebSettings) -> FetchClient:
|
|
"""Create a FetchClient."""
|
|
return FetchClient(settings)
|
|
|
|
|
|
def _make_html(text: str, title: str = "Test Page") -> str:
|
|
"""Create a simple HTML page."""
|
|
return f"<html><head><title>{title}</title></head><body><main><p>{text}</p></main></body></html>"
|
|
|
|
|
|
class TestFetchClientExtraction:
|
|
"""Tests for content extraction in FetchClient."""
|
|
|
|
def test_extract_content_readability(self, client: FetchClient) -> None:
|
|
"""Test that readability extraction works."""
|
|
html = _make_html("This is a test article with enough content to extract.")
|
|
request = FetchRequest(urls=["https://example.com"], skip_validation=True)
|
|
result = client._extract_content(html, "https://example.com", request)
|
|
|
|
assert result["text"]
|
|
assert result["title"] == "Test Page"
|
|
|
|
def test_extract_content_regex_fallback(self, client: FetchClient) -> None:
|
|
"""Test that regex extraction works as last resort."""
|
|
# Minimal HTML that may fail readability/bs4
|
|
html = "<title>Simple</title><div>Hello World</div>"
|
|
request = FetchRequest(urls=["https://example.com"], skip_validation=True)
|
|
result = client._extract_content(html, "https://example.com", request)
|
|
|
|
assert "Hello World" in result["text"] or "Simple" in result.get("title", "")
|
|
|
|
|
|
class TestFetchClientJsDetection:
|
|
"""Tests for JavaScript detection in FetchClient."""
|
|
|
|
def test_detects_enable_javascript_message(self, client: FetchClient) -> None:
|
|
"""Test that 'enable javascript' is detected."""
|
|
html = "<html><body>Please enable JavaScript to continue</body></html>"
|
|
assert client._detect_javascript_required(
|
|
html, "Please enable JavaScript to continue"
|
|
)
|
|
|
|
def test_detects_noscript_warning(self, client: FetchClient) -> None:
|
|
"""Test that noscript with javascript message is detected."""
|
|
html = (
|
|
"<html><body><noscript>You need JavaScript enabled</noscript></body></html>"
|
|
)
|
|
assert client._detect_javascript_required(html, "")
|
|
|
|
def test_no_false_positive_with_content(self, client: FetchClient) -> None:
|
|
"""Test that SPA indicators with sufficient text don't trigger."""
|
|
long_text = "Real content. " * 50
|
|
html = f'<html><body><div id="react-root">{long_text}</div></body></html>'
|
|
assert not client._detect_javascript_required(html, long_text)
|
|
|
|
def test_spa_indicator_with_short_text(self, client: FetchClient) -> None:
|
|
"""Test that SPA indicators with short text trigger detection."""
|
|
html = '<html><body><div id="react-root"></div></body></html>'
|
|
assert client._detect_javascript_required(html, "")
|
|
|
|
|
|
class TestExtractAndAnalyze:
|
|
"""Tests for combined _extract_and_analyze method."""
|
|
|
|
def test_includes_js_detected_key(self, client: FetchClient) -> None:
|
|
"""Test that _extract_and_analyze includes js_detected key."""
|
|
html = _make_html("Normal content for testing purposes.")
|
|
request = FetchRequest(urls=["https://example.com"], skip_validation=True)
|
|
result = client._extract_and_analyze(html, "https://example.com", request)
|
|
|
|
assert "js_detected" in result
|
|
assert isinstance(result["js_detected"], bool)
|
|
|
|
def test_js_detected_true_for_spa(self, client: FetchClient) -> None:
|
|
"""Test that js_detected is True for SPA pages."""
|
|
html = "<html><body><div data-reactroot></div></body></html>"
|
|
request = FetchRequest(urls=["https://example.com"], skip_validation=True)
|
|
result = client._extract_and_analyze(html, "https://example.com", request)
|
|
|
|
assert result["js_detected"] is True
|
|
|
|
|
|
class TestSkipValidation:
|
|
"""Tests for skip_validation flag in FetchClient."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_skip_validation_skips_dns_check(self, client: FetchClient) -> None:
|
|
"""Test that skip_validation=True skips validate_urls_async."""
|
|
request = FetchRequest(
|
|
urls=["https://example.com"],
|
|
skip_validation=True,
|
|
)
|
|
|
|
with patch(
|
|
"web.fetch.client.validate_urls_async", new_callable=AsyncMock
|
|
) as mock_validate:
|
|
with patch.object(
|
|
client, "_fetch_single", new_callable=AsyncMock
|
|
) as mock_fetch:
|
|
mock_fetch.return_value = None
|
|
with contextlib.suppress(Exception):
|
|
await client.fetch(request)
|
|
|
|
mock_validate.assert_not_called()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_no_skip_validation_calls_dns_check(
|
|
self, client: FetchClient
|
|
) -> None:
|
|
"""Test that skip_validation=False calls validate_urls_async."""
|
|
request = FetchRequest(
|
|
urls=["https://example.com"],
|
|
skip_validation=False,
|
|
)
|
|
|
|
with patch(
|
|
"web.fetch.client.validate_urls_async", new_callable=AsyncMock
|
|
) as mock_validate:
|
|
with patch.object(
|
|
client, "_fetch_single", new_callable=AsyncMock
|
|
) as mock_fetch:
|
|
mock_fetch.return_value = None
|
|
with contextlib.suppress(Exception):
|
|
await client.fetch(request)
|
|
|
|
mock_validate.assert_called_once()
|
|
|
|
|
|
class TestHTTP2Enabled:
|
|
"""Tests for HTTP/2 support."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_has_http2_enabled(self, client: FetchClient) -> None:
|
|
"""Test that the underlying httpx client has HTTP/2 enabled."""
|
|
http_client = await client._get_client()
|
|
assert http_client._transport._pool._http2 is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_cleanup(self, client: FetchClient) -> None:
|
|
"""Test that client cleanup works with http2."""
|
|
_ = await client._get_client()
|
|
await client.close()
|
|
assert client._client is None
|
|
|
|
|
|
class TestFetchClientFallbackChain:
|
|
"""Tests for fallback detection in FetchClient."""
|
|
|
|
def test_needs_fallback_on_short_text(self, client: FetchClient) -> None:
|
|
"""Test that short text triggers needs_fallback."""
|
|
html = "<html><body><p>Short</p></body></html>"
|
|
request = FetchRequest(
|
|
urls=["https://example.com"],
|
|
auto_fallback=True,
|
|
min_text_length=200,
|
|
skip_validation=True,
|
|
)
|
|
result = client._extract_and_analyze(html, "https://example.com", request)
|
|
|
|
# Short text should result in needs_fallback in the page result
|
|
assert len(result["text"]) < 200
|