fde4e9f0c6
- Add `prefetch` parameter to CrawlerRunConfig - Add `quick_extract_links()` function for fast link extraction - Add short-circuit in aprocess_html() for prefetch mode - Add 42 tests (unit, integration, regression) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
237 lines
8.4 KiB
Python
237 lines
8.4 KiB
Python
"""Integration tests for prefetch mode with the crawler."""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, BrowserConfig
|
|
|
|
# Use crawl4ai docs as test domain
|
|
TEST_DOMAIN = "https://docs.crawl4ai.com"
|
|
|
|
|
|
class TestPrefetchModeIntegration:
|
|
"""Integration tests for prefetch mode."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_returns_html_and_links(self):
|
|
"""Test that prefetch mode returns HTML and links only."""
|
|
async with AsyncWebCrawler() as crawler:
|
|
config = CrawlerRunConfig(prefetch=True)
|
|
result = await crawler.arun(TEST_DOMAIN, config=config)
|
|
|
|
# Should have HTML
|
|
assert result.html is not None
|
|
assert len(result.html) > 0
|
|
assert "<html" in result.html.lower() or "<!doctype" in result.html.lower()
|
|
|
|
# Should have links
|
|
assert result.links is not None
|
|
assert "internal" in result.links
|
|
assert "external" in result.links
|
|
|
|
# Should NOT have processed content
|
|
assert result.markdown is None or (
|
|
hasattr(result.markdown, 'raw_markdown') and
|
|
result.markdown.raw_markdown is None
|
|
)
|
|
assert result.cleaned_html is None
|
|
assert result.extracted_content is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_preserves_metadata(self):
|
|
"""Test that prefetch mode preserves essential metadata."""
|
|
async with AsyncWebCrawler() as crawler:
|
|
config = CrawlerRunConfig(prefetch=True)
|
|
result = await crawler.arun(TEST_DOMAIN, config=config)
|
|
|
|
# Should have success flag
|
|
assert result.success is True
|
|
|
|
# Should have URL
|
|
assert result.url is not None
|
|
|
|
# Status code should be present
|
|
assert result.status_code is not None or result.status_code == 200
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_with_deep_crawl(self):
|
|
"""Test prefetch mode with deep crawl strategy."""
|
|
from crawl4ai import BFSDeepCrawlStrategy
|
|
|
|
async with AsyncWebCrawler() as crawler:
|
|
config = CrawlerRunConfig(
|
|
prefetch=True,
|
|
deep_crawl_strategy=BFSDeepCrawlStrategy(
|
|
max_depth=1,
|
|
max_pages=3
|
|
)
|
|
)
|
|
|
|
result_container = await crawler.arun(TEST_DOMAIN, config=config)
|
|
|
|
# Handle both list and iterator results
|
|
if hasattr(result_container, '__aiter__'):
|
|
results = [r async for r in result_container]
|
|
else:
|
|
results = list(result_container) if hasattr(result_container, '__iter__') else [result_container]
|
|
|
|
# Each result should have HTML and links
|
|
for result in results:
|
|
assert result.html is not None
|
|
assert result.links is not None
|
|
|
|
# Should have crawled at least one page
|
|
assert len(results) >= 1
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_then_process_with_raw(self):
|
|
"""Test the full two-phase workflow: prefetch then process."""
|
|
async with AsyncWebCrawler() as crawler:
|
|
# Phase 1: Prefetch
|
|
prefetch_config = CrawlerRunConfig(prefetch=True)
|
|
prefetch_result = await crawler.arun(TEST_DOMAIN, config=prefetch_config)
|
|
|
|
stored_html = prefetch_result.html
|
|
|
|
assert stored_html is not None
|
|
assert len(stored_html) > 0
|
|
|
|
# Phase 2: Process with raw: URL
|
|
process_config = CrawlerRunConfig(
|
|
# No prefetch - full processing
|
|
base_url=TEST_DOMAIN # Provide base URL for link resolution
|
|
)
|
|
processed_result = await crawler.arun(
|
|
f"raw:{stored_html}",
|
|
config=process_config
|
|
)
|
|
|
|
# Should now have full processing
|
|
assert processed_result.html is not None
|
|
assert processed_result.success is True
|
|
# Note: cleaned_html and markdown depend on the content
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_links_structure(self):
|
|
"""Test that links have the expected structure."""
|
|
async with AsyncWebCrawler() as crawler:
|
|
config = CrawlerRunConfig(prefetch=True)
|
|
result = await crawler.arun(TEST_DOMAIN, config=config)
|
|
|
|
assert result.links is not None
|
|
|
|
# Check internal links structure
|
|
if result.links["internal"]:
|
|
link = result.links["internal"][0]
|
|
assert "href" in link
|
|
assert "text" in link
|
|
assert link["href"].startswith("http")
|
|
|
|
# Check external links structure (if any)
|
|
if result.links["external"]:
|
|
link = result.links["external"][0]
|
|
assert "href" in link
|
|
assert "text" in link
|
|
assert link["href"].startswith("http")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_config_clone(self):
|
|
"""Test that config.clone() preserves prefetch setting."""
|
|
config = CrawlerRunConfig(prefetch=True)
|
|
cloned = config.clone()
|
|
|
|
assert cloned.prefetch == True
|
|
|
|
# Clone with override
|
|
cloned_false = config.clone(prefetch=False)
|
|
assert cloned_false.prefetch == False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_to_dict(self):
|
|
"""Test that to_dict() includes prefetch."""
|
|
config = CrawlerRunConfig(prefetch=True)
|
|
config_dict = config.to_dict()
|
|
|
|
assert "prefetch" in config_dict
|
|
assert config_dict["prefetch"] == True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_default_false(self):
|
|
"""Test that prefetch defaults to False."""
|
|
config = CrawlerRunConfig()
|
|
assert config.prefetch == False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_explicit_false(self):
|
|
"""Test explicit prefetch=False works like default."""
|
|
async with AsyncWebCrawler() as crawler:
|
|
config = CrawlerRunConfig(prefetch=False)
|
|
result = await crawler.arun(TEST_DOMAIN, config=config)
|
|
|
|
# Should have full processing
|
|
assert result.html is not None
|
|
# cleaned_html should be populated in normal mode
|
|
assert result.cleaned_html is not None
|
|
|
|
|
|
class TestPrefetchPerformance:
|
|
"""Performance-related tests for prefetch mode."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_returns_quickly(self):
|
|
"""Test that prefetch mode returns results quickly."""
|
|
import time
|
|
|
|
async with AsyncWebCrawler() as crawler:
|
|
# Prefetch mode
|
|
start = time.time()
|
|
prefetch_config = CrawlerRunConfig(prefetch=True)
|
|
await crawler.arun(TEST_DOMAIN, config=prefetch_config)
|
|
prefetch_time = time.time() - start
|
|
|
|
# Full mode
|
|
start = time.time()
|
|
full_config = CrawlerRunConfig()
|
|
await crawler.arun(TEST_DOMAIN, config=full_config)
|
|
full_time = time.time() - start
|
|
|
|
# Log times for debugging
|
|
print(f"\nPrefetch: {prefetch_time:.3f}s, Full: {full_time:.3f}s")
|
|
|
|
# Prefetch should not be significantly slower
|
|
# (may be same or slightly faster depending on content)
|
|
# This is a soft check - mostly for logging
|
|
|
|
|
|
class TestPrefetchWithRawHTML:
|
|
"""Test prefetch mode with raw HTML input."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_prefetch_with_raw_html(self):
|
|
"""Test prefetch mode works with raw: URL scheme."""
|
|
sample_html = """
|
|
<html>
|
|
<head><title>Test Page</title></head>
|
|
<body>
|
|
<h1>Hello World</h1>
|
|
<a href="/link1">Link 1</a>
|
|
<a href="/link2">Link 2</a>
|
|
<a href="https://external.com/page">External</a>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
async with AsyncWebCrawler() as crawler:
|
|
config = CrawlerRunConfig(
|
|
prefetch=True,
|
|
base_url="https://example.com"
|
|
)
|
|
result = await crawler.arun(f"raw:{sample_html}", config=config)
|
|
|
|
assert result.success is True
|
|
assert result.html is not None
|
|
assert result.links is not None
|
|
|
|
# Should have extracted links
|
|
assert len(result.links["internal"]) >= 2
|
|
assert len(result.links["external"]) >= 1
|