Join our core engineering team...
Drive our growth initiatives...
This domain is for use in documentation examples without needing permission. Avoid use in operations.
Learn more
Testing Google Analytics integration
Welcome to the documentation
This is a test paragraph.
- 
- 
-
- '''
-
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- exclude_external_images=True
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun(f"raw://{html_with_images}", config=config)
- assert result.success
- # External images should be excluded
- assert "external.com" not in result.cleaned_html
-
- @pytest.mark.asyncio
- async def test_llm_extraction_strategy_fix(self):
- """Test LLM extraction strategy choices error fix"""
- if not os.getenv("OPENAI_API_KEY"):
- pytest.skip("OpenAI API key not available")
-
- llm_config = LLMConfig(
- provider="openai/gpt-4o-mini",
- api_token=os.getenv("OPENAI_API_KEY")
- )
-
- strategy = LLMExtractionStrategy(
- llm_config=llm_config,
- instruction="Extract the main heading",
- extraction_type="block"
- )
-
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- extraction_strategy=strategy
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
- # Should not throw 'str' object has no attribute 'choices' error
- assert result.extracted_content is not None
-
- @pytest.mark.asyncio
- async def test_wait_for_timeout(self):
- """Test separate timeout for wait_for condition"""
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- wait_for="css:non-existent-element",
- wait_for_timeout=1000 # 1 second timeout
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- # Should timeout gracefully and still return result
- assert result is not None
-
- @pytest.mark.asyncio
- async def test_bm25_content_filter_language_parameter(self):
- """Test BM25 filter with language parameter for stemming"""
- content_filter = BM25ContentFilter(
- user_query="test content",
- language="english",
- use_stemming=True
- )
-
- markdown_generator = DefaultMarkdownGenerator(
- content_filter=content_filter
- )
-
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- markdown_generator=markdown_generator
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
- assert result.markdown is not None
-
- @pytest.mark.asyncio
- async def test_url_normalization(self):
- """Test URL normalization for invalid schemes and trailing slashes"""
- async with AsyncWebCrawler() as crawler:
- # Test with trailing slash
- result = await crawler.arun(
- "https://httpbin.org/html/",
- config=CrawlerRunConfig(cache_mode=CacheMode.BYPASS)
- )
- assert result.success
-
- @pytest.mark.asyncio
- async def test_max_scroll_steps(self):
- """Test max_scroll_steps parameter for full page scanning"""
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- scan_full_page=True,
- max_scroll_steps=3
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
-
- @pytest.mark.asyncio
- async def test_async_url_seeder(self):
- """Test AsyncUrlSeeder functionality"""
- seeder = AsyncUrlSeeder(
- base_url="https://httpbin.org",
- max_depth=1,
- max_urls=5
- )
-
- async with AsyncWebCrawler() as crawler:
- urls = await seeder.seed(crawler)
- assert isinstance(urls, list)
- assert len(urls) <= 5
-
- @pytest.mark.asyncio
- async def test_pdf_processing_timeout(self):
- """Test PDF processing with timeout"""
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- pdf=True,
- pdf_timeout=10000 # 10 seconds
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
- # PDF might be None for HTML pages, but should not hang
- assert result.pdf is not None or result.pdf is None
-
- @pytest.mark.asyncio
- async def test_browser_session_management(self):
- """Test improved browser session management"""
- browser_config = BrowserConfig(
- headless=True,
- use_persistent_context=True
- )
-
- async with AsyncWebCrawler(config=browser_config) as crawler:
- result = await crawler.arun(
- "https://httpbin.org/html",
- config=CrawlerRunConfig(cache_mode=CacheMode.BYPASS)
- )
- assert result.success
-
- @pytest.mark.asyncio
- async def test_memory_management(self):
- """Test memory management features"""
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- memory_threshold_percent=80.0,
- check_interval=1.0,
- memory_wait_timeout=600 # 10 minutes default
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
-
- @pytest.mark.asyncio
- async def test_virtual_scroll_support(self):
- """Test virtual scroll support for modern web scraping"""
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- scan_full_page=True,
- virtual_scroll=True
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
-
- @pytest.mark.asyncio
- async def test_adaptive_crawling(self):
- """Test adaptive crawling feature"""
- config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- adaptive_crawling=True
- )
-
- async with AsyncWebCrawler() as crawler:
- result = await crawler.arun("https://httpbin.org/html", config=config)
- assert result.success
-
-
-if __name__ == "__main__":
- # Run the tests
- pytest.main([__file__, "-v"])
diff --git a/tests/test_arun_many.py b/tests/test_arun_many.py
deleted file mode 100644
index 2a315a2a3..000000000
--- a/tests/test_arun_many.py
+++ /dev/null
@@ -1,42 +0,0 @@
-"""
-Test example for multiple crawler configs feature
-"""
-import asyncio
-import sys
-from pathlib import Path
-
-# Add parent directory to path for imports
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, CacheMode
-from crawl4ai.processors.pdf import PDFContentScrapingStrategy
-
-
-async def test_run_many():
- default_config = CrawlerRunConfig(
- cache_mode=CacheMode.BYPASS,
- # scraping_strategy=PDFContentScrapingStrategy()
- )
-
- test_urls = [
- # "https://blog.python.org/", # Blog URL
- "https://www.python.org/", # Generic HTTPS page
- "https://www.kidocode.com/", # Generic HTTPS page
- "https://www.example.com/", # Generic HTTPS page
- # "https://www.w3.org/WAI/ER/tests/xhtml/testfiles/resources/pdf/dummy.pdf",
- ]
-
- async with AsyncWebCrawler() as crawler:
- # Single config - traditional usage still works
- print("Test 1: Single config (backwards compatible)")
- result = await crawler.arun_many(
- urls=test_urls[:2],
- config=default_config
- )
- print(f"Crawled {len(result)} URLs with single config\n")
- for item in result:
- print(f" {item.url} -> {item.status_code}")
-
-
-if __name__ == "__main__":
- asyncio.run(test_run_many())
diff --git a/tests/general/test_async_webcrawler.py b/tests/test_async_web_crawler.py
similarity index 64%
rename from tests/general/test_async_webcrawler.py
rename to tests/test_async_web_crawler.py
index 4d7aa815f..895944b04 100644
--- a/tests/general/test_async_webcrawler.py
+++ b/tests/test_async_web_crawler.py
@@ -1,14 +1,41 @@
-import asyncio
import pytest
-from typing import List
+
from crawl4ai import (
AsyncWebCrawler,
- BrowserConfig,
+ BrowserConfig,
CrawlerRunConfig,
MemoryAdaptiveDispatcher,
RateLimiter,
- CacheMode
)
+from tests.helpers import EXAMPLE_URL
+
+
+@pytest.mark.asyncio
+async def test_arun():
+ async with AsyncWebCrawler() as crawler:
+ result = await crawler.arun(
+ url=EXAMPLE_URL
+ )
+ assert result.status_code == 200
+ assert result.url == EXAMPLE_URL
+ assert result.markdown is not None
+
+@pytest.mark.asyncio
+async def test_arun_many():
+ test_urls = [
+ "https://www.python.org/",
+ EXAMPLE_URL,
+ ]
+
+ async with AsyncWebCrawler() as crawler:
+ results = await crawler.arun_many(
+ urls=test_urls[:2],
+ )
+ assert len(results) == len(test_urls)
+ for item in results:
+ assert item.status_code == 200
+ assert item.markdown is not None
+ assert item.url in test_urls
@pytest.mark.asyncio
@pytest.mark.parametrize("viewport", [
@@ -28,9 +55,8 @@ async def test_viewport_config(viewport):
async with AsyncWebCrawler(config=browser_config) as crawler:
result = await crawler.arun(
- url="https://example.com",
+ url=EXAMPLE_URL,
config=CrawlerRunConfig(
- # cache_mode=CacheMode.BYPASS,
page_timeout=30000 # 30 seconds
)
)
@@ -52,7 +78,7 @@ async def test_memory_management():
max_session_permit=5
)
- urls = ["https://example.com"] * 3 # Test with multiple identical URLs
+ urls = [EXAMPLE_URL] * 3 # Test with multiple identical URLs
async with AsyncWebCrawler(config=browser_config) as crawler:
results = await crawler.arun_many(
@@ -80,7 +106,7 @@ async def test_rate_limiting():
)
urls = [
- "https://example.com",
+ EXAMPLE_URL,
"https://example.org",
"https://example.net"
]
@@ -109,13 +135,14 @@ async def test_javascript_execution():
async with AsyncWebCrawler(config=browser_config) as crawler:
result = await crawler.arun(
- url="https://example.com",
+ url=EXAMPLE_URL,
config=CrawlerRunConfig(
js_code=js_code,
page_timeout=30000
)
)
- assert result.success
+
+ assert result.success
@pytest.mark.asyncio
@pytest.mark.parametrize("error_url", [
@@ -135,15 +162,35 @@ async def test_error_handling(error_url):
result = await crawler.arun(
url=error_url,
config=CrawlerRunConfig(
- page_timeout=10000, # Short timeout for error cases
- cache_mode=CacheMode.BYPASS
+ page_timeout=10000
)
)
- assert not result.success
- assert result.error_message is not None
-
-if __name__ == "__main__":
- asyncio.run(test_viewport_config((1024, 768)))
- asyncio.run(test_memory_management())
- asyncio.run(test_rate_limiting())
- asyncio.run(test_javascript_execution())
\ No newline at end of file
+
+ assert not result.success
+ assert result.error_message is not None
+
+
+@pytest.mark.asyncio
+async def test_extract_media():
+ async with AsyncWebCrawler() as crawler:
+ url = "https://www.nbcnews.com/business"
+ result = await crawler.arun(url=url)
+
+ assert result.success
+ assert result.media
+ assert result.media["images"]
+ assert any(img["src"] for img in result.media["images"])
+ assert any(img["alt"] for img in result.media["images"])
+ assert any(img["score"] for img in result.media["images"])
+
+@pytest.mark.asyncio
+async def test_extract_metadata():
+ async with AsyncWebCrawler() as crawler:
+ url = "https://www.nbcnews.com/business"
+ result = await crawler.arun(url=url)
+
+ assert result.success
+ assert result.metadata
+ assert all(
+ key in result.metadata for key in ["title", "description", "keywords"]
+ )
diff --git a/tests/test_caching.py b/tests/test_caching.py
new file mode 100644
index 000000000..7d10f0f63
--- /dev/null
+++ b/tests/test_caching.py
@@ -0,0 +1,146 @@
+import pytest
+
+from unittest.mock import Mock, patch
+
+import pytest_asyncio
+
+from crawl4ai.async_configs import CrawlerRunConfig
+from crawl4ai.async_webcrawler import AsyncWebCrawler
+from crawl4ai.cache_context import CacheMode
+from crawl4ai.models import AsyncCrawlResponse
+from tests.helpers import EXAMPLE_RAW_HTML, EXAMPLE_URL, TestCacheClient
+
+
+@pytest_asyncio.fixture
+async def mock_async_crawl_response(monkeypatch):
+ mock_crawl_response = AsyncCrawlResponse(
+ html=EXAMPLE_RAW_HTML,
+ response_headers={
+ 'accept-ranges': 'bytes',
+ 'alt-svc': 'h3=":443"; ma=93600',
+ 'cache-control': 'max-age=86000',
+ 'content-length': '513',
+ 'content-type': 'text/html',
+ 'date': 'Wed, 19 Nov 2025 20:09:52 GMT',
+ 'etag': '"bc2473a18e003bdb249eba5ce893033f:1760028122.592274"',
+ 'last-modified': 'Thu, 09 Oct 2025 16:42:02 GMT'
+ },
+ js_execution_result=None,
+ status_code=200,
+ screenshot=None,
+ pdf_data=None,
+ mhtml_data=None,
+ downloaded_files=None,
+ ssl_certificate=None,
+ redirected_url=EXAMPLE_URL,
+ network_requests=None,
+ console_messages=None
+ )
+ async def mock_crawl(self, url, **kwargs):
+ return mock_crawl_response
+ monkeypatch.setattr("crawl4ai.async_crawler_strategy.AsyncPlaywrightCrawlerStrategy.crawl", mock_crawl)
+
+
+@pytest.mark.asyncio
+async def test_caching():
+ cache_client = TestCacheClient()
+
+ async with AsyncWebCrawler(cache_client=cache_client) as crawler:
+ # First crawl (should not use cache)
+ result1 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED
+ ))
+ cache_size = cache_client.count()
+
+ assert result1.success
+ assert cache_size == 1
+
+ # Second crawl (should use cache)
+ result2 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED
+ ))
+ final_cache_size = cache_client.count()
+
+ assert result2.success
+ assert result2.html == result1.html
+ assert final_cache_size == 1
+
+ cache_client.cleanup()
+
+
+@pytest.mark.asyncio
+async def test_cache_excluded_tags(mock_async_crawl_response):
+ cache_client = TestCacheClient()
+
+ async with AsyncWebCrawler(cache_client=cache_client) as crawler:
+ result1 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED,
+ excluded_tags=["p"]
+ ))
+ cache_size = cache_client.count()
+
+ assert result1.success
+ assert result1.markdown == "# Example Domain\n"
+ assert cache_size == 1
+
+ result2 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED,
+ excluded_tags=["h1"]
+ ))
+ cache_size2 = cache_client.count()
+
+ assert result2.success
+ assert result2.markdown == "This domain is for use in documentation examples without needing permission. Avoid use in operations.\n[Learn more](https://iana.org/domains/example)\n"
+ assert cache_size2 == cache_size
+
+ result3 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED,
+ ))
+ cache_size3 = cache_client.count()
+
+ assert result3.success
+ assert result3.markdown == "# Example Domain\nThis domain is for use in documentation examples without needing permission. Avoid use in operations.\n[Learn more](https://iana.org/domains/example)\n"
+ assert cache_size3 == cache_size
+
+ cache_client.cleanup()
+
+
+@pytest.mark.asyncio
+async def test_bypass_cache():
+ cache_client = TestCacheClient()
+ get_spy = Mock(wraps=cache_client.get)
+
+ with patch.object(TestCacheClient, 'get', get_spy):
+ async with AsyncWebCrawler(cache_client=cache_client) as crawler:
+ result1 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED
+ ))
+
+ assert result1.success
+ assert get_spy.call_count == 1
+
+ result2 = await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.BYPASS
+ ))
+
+ assert result2.success
+ assert get_spy.call_count == 1
+
+ cache_client.cleanup()
+
+
+@pytest.mark.asyncio
+async def test_clear_cache():
+ cache_client = TestCacheClient()
+
+ async with AsyncWebCrawler(cache_client=cache_client) as crawler:
+ await crawler.arun(url=EXAMPLE_URL, config=CrawlerRunConfig(
+ cache_mode=CacheMode.ENABLED
+ ))
+ assert cache_client.count() == 1
+
+ cache_client.clear(prefix="")
+ assert cache_client.count() == 0
+
+ cache_client.cleanup()
+
diff --git a/tests/test_cli_docs.py b/tests/test_cli_docs.py
deleted file mode 100644
index 6941f20db..000000000
--- a/tests/test_cli_docs.py
+++ /dev/null
@@ -1,44 +0,0 @@
-import asyncio
-from crawl4ai.docs_manager import DocsManager
-from click.testing import CliRunner
-from crawl4ai.cli import cli
-
-
-def test_cli():
- """Test all CLI commands"""
- runner = CliRunner()
-
- print("\n1. Testing docs update...")
- # Use sync version for testing
- docs_manager = DocsManager()
- loop = asyncio.get_event_loop()
- loop.run_until_complete(docs_manager.fetch_docs())
-
- # print("\n2. Testing listing...")
- # result = runner.invoke(cli, ['docs', 'list'])
- # print(f"Status: {'โ
' if result.exit_code == 0 else 'โ'}")
- # print(result.output)
-
- # print("\n2. Testing index building...")
- # result = runner.invoke(cli, ['docs', 'index'])
- # print(f"Status: {'โ
' if result.exit_code == 0 else 'โ'}")
- # print(f"Output: {result.output}")
-
- # print("\n3. Testing search...")
- # result = runner.invoke(cli, ['docs', 'search', 'how to use crawler', '--build-index'])
- # print(f"Status: {'โ
' if result.exit_code == 0 else 'โ'}")
- # print(f"First 200 chars: {result.output[:200]}...")
-
- # print("\n4. Testing combine with sections...")
- # result = runner.invoke(cli, ['docs', 'combine', 'chunking_strategies', 'extraction_strategies', '--mode', 'extended'])
- # print(f"Status: {'โ
' if result.exit_code == 0 else 'โ'}")
- # print(f"First 200 chars: {result.output[:200]}...")
-
- print("\n5. Testing combine all sections...")
- result = runner.invoke(cli, ["docs", "combine", "--mode", "condensed"])
- print(f"Status: {'โ
' if result.exit_code == 0 else 'โ'}")
- print(f"First 200 chars: {result.output[:200]}...")
-
-
-if __name__ == "__main__":
- test_cli()
diff --git a/tests/test_config_matching_only.py b/tests/test_config_matching_only.py
deleted file mode 100644
index 0f21666e5..000000000
--- a/tests/test_config_matching_only.py
+++ /dev/null
@@ -1,131 +0,0 @@
-"""
-Test only the config matching logic without running crawler
-"""
-import sys
-from pathlib import Path
-
-# Add parent directory to path for imports
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from crawl4ai.async_configs import CrawlerRunConfig, MatchMode
-
-def test_all_matching_scenarios():
- print("Testing CrawlerRunConfig.is_match() method")
- print("=" * 50)
-
- # Test 1: Single string pattern
- print("\n1. Single string pattern (glob style)")
- config = CrawlerRunConfig(
- url_matcher="*.pdf",
- # For example we can set this => scraping_strategy=PDFContentScrapingStrategy()
- )
- test_urls = [
- ("https://example.com/file.pdf", True),
- ("https://example.com/doc.PDF", False), # Case sensitive
- ("https://example.com/file.txt", False),
- ("file.pdf", True),
- ]
- for url, expected in test_urls:
- result = config.is_match(url)
- status = "โ" if result == expected else "โ"
- print(f" {status} {url} -> {result}")
-
- # Test 2: List of patterns with OR
- print("\n2. List of patterns with OR (default)")
- config = CrawlerRunConfig(
- url_matcher=["*/article/*", "*/blog/*", "*.html"],
- match_mode=MatchMode.OR
- )
- test_urls = [
- ("https://example.com/article/news", True),
- ("https://example.com/blog/post", True),
- ("https://example.com/page.html", True),
- ("https://example.com/page.php", False),
- ]
- for url, expected in test_urls:
- result = config.is_match(url)
- status = "โ" if result == expected else "โ"
- print(f" {status} {url} -> {result}")
-
- # Test 3: Custom function
- print("\n3. Custom function matcher")
- config = CrawlerRunConfig(
- url_matcher=lambda url: 'api' in url and (url.endswith('.json') or url.endswith('.xml'))
- )
- test_urls = [
- ("https://api.example.com/data.json", True),
- ("https://api.example.com/data.xml", True),
- ("https://api.example.com/data.html", False),
- ("https://example.com/data.json", False), # No 'api'
- ]
- for url, expected in test_urls:
- result = config.is_match(url)
- status = "โ" if result == expected else "โ"
- print(f" {status} {url} -> {result}")
-
- # Test 4: Mixed list with AND
- print("\n4. Mixed patterns and functions with AND")
- config = CrawlerRunConfig(
- url_matcher=[
- "https://*", # Must be HTTPS
- lambda url: '.com' in url, # Must have .com
- lambda url: len(url) < 50 # Must be short
- ],
- match_mode=MatchMode.AND
- )
- test_urls = [
- ("https://example.com/page", True),
- ("http://example.com/page", False), # Not HTTPS
- ("https://example.org/page", False), # No .com
- ("https://example.com/" + "x" * 50, False), # Too long
- ]
- for url, expected in test_urls:
- result = config.is_match(url)
- status = "โ" if result == expected else "โ"
- print(f" {status} {url} -> {result}")
-
- # Test 5: Complex real-world scenario
- print("\n5. Complex pattern combinations")
- config = CrawlerRunConfig(
- url_matcher=[
- "*/api/v[0-9]/*", # API versioned endpoints
- lambda url: 'graphql' in url, # GraphQL endpoints
- "*.json" # JSON files
- ],
- match_mode=MatchMode.OR
- )
- test_urls = [
- ("https://example.com/api/v1/users", True),
- ("https://example.com/api/v2/posts", True),
- ("https://example.com/graphql", True),
- ("https://example.com/data.json", True),
- ("https://example.com/api/users", False), # No version
- ]
- for url, expected in test_urls:
- result = config.is_match(url)
- status = "โ" if result == expected else "โ"
- print(f" {status} {url} -> {result}")
-
- # Test 6: Edge cases
- print("\n6. Edge cases")
-
- # No matcher
- config = CrawlerRunConfig()
- result = config.is_match("https://example.com")
- print(f" {'โ' if not result else 'โ'} No matcher -> {result}")
-
- # Empty list
- config = CrawlerRunConfig(url_matcher=[])
- result = config.is_match("https://example.com")
- print(f" {'โ' if not result else 'โ'} Empty list -> {result}")
-
- # None in list (should be skipped)
- config = CrawlerRunConfig(url_matcher=["*.pdf", None, "*.doc"])
- result = config.is_match("test.pdf")
- print(f" {'โ' if result else 'โ'} List with None -> {result}")
-
- print("\n" + "=" * 50)
- print("All matching tests completed!")
-
-if __name__ == "__main__":
- test_all_matching_scenarios()
\ No newline at end of file
diff --git a/tests/test_config_selection.py b/tests/test_config_selection.py
deleted file mode 100644
index 97245f9f1..000000000
--- a/tests/test_config_selection.py
+++ /dev/null
@@ -1,87 +0,0 @@
-"""
-Test config selection logic in dispatchers
-"""
-import asyncio
-import sys
-from pathlib import Path
-from unittest.mock import AsyncMock, MagicMock
-
-# Add parent directory to path for imports
-sys.path.insert(0, str(Path(__file__).parent.parent))
-
-from crawl4ai.async_configs import CrawlerRunConfig, MatchMode
-from crawl4ai.async_dispatcher import BaseDispatcher, MemoryAdaptiveDispatcher
-
-class TestDispatcher(BaseDispatcher):
- """Simple test dispatcher to verify config selection"""
-
- async def crawl_url(self, url, config, task_id, **kwargs):
- # Just return which config was selected
- selected = self.select_config(url, config)
- return {"url": url, "config_id": id(selected)}
-
- async def run_urls(self, urls, crawler, config):
- results = []
- for url in urls:
- result = await self.crawl_url(url, config, "test")
- results.append(result)
- return results
-
-async def test_dispatcher_config_selection():
- print("Testing dispatcher config selection")
- print("=" * 50)
-
- # Create test configs with different matchers
- pdf_config = CrawlerRunConfig(url_matcher="*.pdf")
- api_config = CrawlerRunConfig(url_matcher=lambda url: 'api' in url)
- default_config = CrawlerRunConfig() # No matcher
-
- configs = [pdf_config, api_config, default_config]
-
- # Create test dispatcher
- dispatcher = TestDispatcher()
-
- # Test single config
- print("\nTest 1: Single config")
- result = await dispatcher.crawl_url("https://example.com/file.pdf", pdf_config, "test1")
- assert result["config_id"] == id(pdf_config)
- print("โ Single config works")
-
- # Test config list selection
- print("\nTest 2: Config list selection")
- test_cases = [
- ("https://example.com/file.pdf", id(pdf_config)),
- ("https://api.example.com/data", id(api_config)),
- ("https://example.com/page", id(configs[0])), # No match, uses first
- ]
-
- for url, expected_id in test_cases:
- result = await dispatcher.crawl_url(url, configs, "test")
- assert result["config_id"] == expected_id, f"URL {url} got wrong config"
- print(f"โ {url} -> correct config selected")
-
- # Test with MemoryAdaptiveDispatcher
- print("\nTest 3: MemoryAdaptiveDispatcher config selection")
- mem_dispatcher = MemoryAdaptiveDispatcher()
-
- # Test select_config method directly
- selected = mem_dispatcher.select_config("https://example.com/doc.pdf", configs)
- assert selected == pdf_config
- print("โ MemoryAdaptiveDispatcher.select_config works")
-
- # Test empty config list
- print("\nTest 4: Edge cases")
- selected = mem_dispatcher.select_config("https://example.com", [])
- assert isinstance(selected, CrawlerRunConfig) # Should return default
- print("โ Empty config list returns default config")
-
- # Test None config
- selected = mem_dispatcher.select_config("https://example.com", None)
- assert isinstance(selected, CrawlerRunConfig) # Should return default
- print("โ None config returns default config")
-
- print("\n" + "=" * 50)
- print("All dispatcher tests passed! โ")
-
-if __name__ == "__main__":
- asyncio.run(test_dispatcher_config_selection())
\ No newline at end of file
diff --git a/tests/test_crawler_run_config.py b/tests/test_crawler_run_config.py
new file mode 100644
index 000000000..bc550c036
--- /dev/null
+++ b/tests/test_crawler_run_config.py
@@ -0,0 +1,165 @@
+import pytest
+
+from crawl4ai import AsyncWebCrawler, CrawlerRunConfig, LinkPreviewConfig
+from crawl4ai.async_configs import MatchMode
+from tests.helpers import EXAMPLE_URL
+
+
+def test_is_match_single_pattern():
+ config = CrawlerRunConfig(
+ url_matcher="*.pdf"
+ )
+ test_urls = [
+ (f"{EXAMPLE_URL}/file.pdf", True),
+ (f"{EXAMPLE_URL}/doc.PDF", False), # Case sensitive
+ (f"{EXAMPLE_URL}/file.txt", False),
+ ("file.pdf", True),
+ ]
+
+ for url, expected in test_urls:
+ assert config.is_match(url) == expected
+
+def test_is_match_with_or_mode():
+ config = CrawlerRunConfig(
+ url_matcher=["*/article/*", "*/blog/*", "*.html"],
+ match_mode=MatchMode.OR
+ )
+ test_urls = [
+ (f"{EXAMPLE_URL}/article/news", True),
+ (f"{EXAMPLE_URL}/blog/post", True),
+ (f"{EXAMPLE_URL}/page.html", True),
+ (f"{EXAMPLE_URL}/page.php", False),
+ ]
+ for url, expected in test_urls:
+ assert config.is_match(url) == expected
+
+def test_is_match_custom_function():
+ config = CrawlerRunConfig(
+ url_matcher=lambda url: 'api' in url and (url.endswith('.json') or url.endswith('.xml'))
+ )
+ test_urls = [
+ (f"{EXAMPLE_URL}/api/data.json", True),
+ (f"{EXAMPLE_URL}/api/data.xml", True),
+ (f"{EXAMPLE_URL}/data.html", False),
+ (f"{EXAMPLE_URL}/data.json", False), # No 'api'
+ ]
+ for url, expected in test_urls:
+ assert config.is_match(url) == expected
+
+def test_is_match_mixed_with_and_mode():
+ config = CrawlerRunConfig(
+ url_matcher=[
+ "https://*", # Must be HTTPS
+ lambda url: '.com' in url, # Must have .com
+ lambda url: len(url) < 50 # Must be short
+ ],
+ match_mode=MatchMode.AND
+ )
+ test_urls = [
+ (f"{EXAMPLE_URL}/page", True),
+ ("http://example.com/page", False), # Not HTTPS
+ ("example.org/page", False), # No .com
+ (f"{EXAMPLE_URL}/" + "x" * 50, False), # Too long
+ ]
+ for url, expected in test_urls:
+ assert config.is_match(url) == expected
+
+def test_is_match_mixed_with_or_mode():
+ config = CrawlerRunConfig(
+ url_matcher=[
+ "*/api/v[0-9]/*", # API versioned endpoints
+ lambda url: 'graphql' in url, # GraphQL endpoints
+ "*.json" # JSON files
+ ],
+ match_mode=MatchMode.OR
+ )
+ test_urls = [
+ (f"{EXAMPLE_URL}/api/v1/users", True),
+ (f"{EXAMPLE_URL}/api/v2/posts", True),
+ (f"{EXAMPLE_URL}/graphql", True),
+ (f"{EXAMPLE_URL}/data.json", True),
+ (f"{EXAMPLE_URL}/api/users", False), # No version
+ ]
+ for url, expected in test_urls:
+ assert config.is_match(url) == expected
+
+def test_is_match_no_matcher():
+ config = CrawlerRunConfig()
+ assert config.is_match(EXAMPLE_URL)
+
+def test_is_match_empty_list():
+ config = CrawlerRunConfig(url_matcher=[])
+ assert not config.is_match(EXAMPLE_URL)
+
+@pytest.mark.asyncio
+async def test_link_preview():
+ config = CrawlerRunConfig(
+ link_preview_config=LinkPreviewConfig(),
+ score_links=True,
+ exclude_external_links=True,
+ )
+ test_url = "https://docs.python.org/3/"
+
+ async with AsyncWebCrawler() as crawler:
+ result = await crawler.arun(test_url, config=config)
+
+ assert result.success
+ assert result.links
+ assert result.links["external"] == []
+
+ internal_links = result.links["internal"]
+
+ assert any(il["total_score"] for il in internal_links)
+ assert all(il["href"] for il in internal_links)
+ assert all("title" in il for il in internal_links)
+ assert all("text" in il for il in internal_links)
+ assert all("head_data" in il for il in internal_links)
+ assert all("head_extraction_status" in il for il in internal_links)
+ assert all("head_extraction_error" in il for il in internal_links)
+ assert all("total_score" in il for il in internal_links)
+
+@pytest.mark.asyncio
+async def test_css_selector():
+ async with AsyncWebCrawler() as crawler:
+ result = await crawler.arun(
+ url=EXAMPLE_URL, config=CrawlerRunConfig(
+ css_selector="h1, title"
+ )
+ )
+
+ assert result.success
+ assert (
+ "