366 lines
15 KiB
Python
366 lines
15 KiB
Python
|
#!/usr/bin/env python3
|
|||
|
|
"""
|
|||
|
|
Real-world testing of Crawailer JavaScript API enhancements.
|
|||
|
|
Tests various website types to validate production readiness.
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
import sys
|
|||
|
|
import time
|
|||
|
|
from datetime import datetime
|
|||
|
|
from typing import List, Dict, Any
|
|||
|
|
|
|||
|
|
# Add src to path to use our enhanced implementation
|
|||
|
|
sys.path.insert(0, 'src')
|
|||
|
|
|
|||
|
|
import crawailer as web
|
|||
|
|
|
|||
|
|
|
|||
|
|
class RealWorldTester:
|
|||
|
|
"""Test suite for real-world website crawling with JavaScript enhancement."""
|
|||
|
|
|
|||
|
|
def __init__(self):
|
|||
|
|
self.results = []
|
|||
|
|
self.test_start_time = None
|
|||
|
|
|
|||
|
|
async def test_static_content_baseline(self):
|
|||
|
|
"""Test with static content to ensure basic functionality works."""
|
|||
|
|
print("🧪 Testing Static Content (Baseline)")
|
|||
|
|
print("-" * 50)
|
|||
|
|
|
|||
|
|
test_cases = [
|
|||
|
|
{
|
|||
|
|
"name": "Wikipedia Article",
|
|||
|
|
"url": "https://en.wikipedia.org/wiki/Web_scraping",
|
|||
|
|
"expected_elements": ["Web scraping", "content", "extraction"],
|
|||
|
|
"use_js": False
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"name": "Example.com",
|
|||
|
|
"url": "https://example.com",
|
|||
|
|
"expected_elements": ["Example Domain", "information", "examples"],
|
|||
|
|
"use_js": False
|
|||
|
|
}
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
for test in test_cases:
|
|||
|
|
await self._run_test_case(test)
|
|||
|
|
|
|||
|
|
async def test_dynamic_content_scenarios(self):
|
|||
|
|
"""Test JavaScript-enhanced content extraction."""
|
|||
|
|
print("\n🚀 Testing Dynamic Content with JavaScript")
|
|||
|
|
print("-" * 50)
|
|||
|
|
|
|||
|
|
test_cases = [
|
|||
|
|
{
|
|||
|
|
"name": "GitHub Repository (Dynamic Loading)",
|
|||
|
|
"url": "https://github.com/microsoft/playwright",
|
|||
|
|
"script": """
|
|||
|
|
// Wait for dynamic content and return repository stats
|
|||
|
|
await new Promise(r => setTimeout(r, 2000));
|
|||
|
|
const stars = document.querySelector('[data-view-component="true"] strong')?.innerText || 'unknown';
|
|||
|
|
return {stars: stars, loaded: true};
|
|||
|
|
""",
|
|||
|
|
"expected_elements": ["Playwright", "browser", "automation"],
|
|||
|
|
"use_js": True
|
|||
|
|
},
|
|||
|
|
{
|
|||
|
|
"name": "JSONPlaceholder API Demo",
|
|||
|
|
"url": "https://jsonplaceholder.typicode.com/",
|
|||
|
|
"script": """
|
|||
|
|
// Look for API endpoints and examples
|
|||
|
|
const links = Array.from(document.querySelectorAll('a')).map(a => a.href);
|
|||
|
|
const codeBlocks = Array.from(document.querySelectorAll('code')).map(c => c.innerText);
|
|||
|
|
return {
|
|||
|
|
links_found: links.length,
|
|||
|
|
code_examples: codeBlocks.length,
|
|||
|
|
has_api_info: document.body.innerText.includes('REST API')
|
|||
|
|
};
|
|||
|
|
""",
|
|||
|
|
"expected_elements": ["REST API", "JSON", "placeholder"],
|
|||
|
|
"use_js": True
|
|||
|
|
}
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
for test in test_cases:
|
|||
|
|
await self._run_test_case(test)
|
|||
|
|
|
|||
|
|
async def test_spa_and_modern_sites(self):
|
|||
|
|
"""Test Single Page Applications and modern JavaScript-heavy sites."""
|
|||
|
|
print("\n⚡ Testing SPAs and Modern JavaScript Sites")
|
|||
|
|
print("-" * 50)
|
|||
|
|
|
|||
|
|
test_cases = [
|
|||
|
|
{
|
|||
|
|
"name": "React Documentation",
|
|||
|
|
"url": "https://react.dev/",
|
|||
|
|
"script": """
|
|||
|
|
// Wait for React app to load
|
|||
|
|
await new Promise(r => setTimeout(r, 3000));
|
|||
|
|
const title = document.querySelector('h1')?.innerText || 'No title found';
|
|||
|
|
const navItems = document.querySelectorAll('nav a').length;
|
|||
|
|
return {
|
|||
|
|
page_title: title,
|
|||
|
|
navigation_items: navItems,
|
|||
|
|
react_loaded: !!window.React || document.body.innerText.includes('React')
|
|||
|
|
};
|
|||
|
|
""",
|
|||
|
|
"expected_elements": ["React", "JavaScript", "library"],
|
|||
|
|
"use_js": True
|
|||
|
|
}
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
for test in test_cases:
|
|||
|
|
await self._run_test_case(test)
|
|||
|
|
|
|||
|
|
async def test_batch_processing(self):
|
|||
|
|
"""Test get_many() with multiple sites and different JavaScript requirements."""
|
|||
|
|
print("\n📦 Testing Batch Processing with Mixed JavaScript")
|
|||
|
|
print("-" * 50)
|
|||
|
|
|
|||
|
|
urls = [
|
|||
|
|
"https://httpbin.org/html", # Static HTML
|
|||
|
|
"https://httpbin.org/json", # JSON endpoint
|
|||
|
|
"https://example.com" # Simple static page
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
scripts = [
|
|||
|
|
"document.querySelector('h1')?.innerText || 'No H1 found'", # Extract title
|
|||
|
|
"JSON.stringify(Object.keys(window).slice(0, 5))", # Get some window properties
|
|||
|
|
None # No script for simple page
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
start_time = time.time()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
print(f"Processing {len(urls)} URLs with mixed JavaScript requirements...")
|
|||
|
|
|
|||
|
|
results = await web.get_many(urls, script=scripts, max_concurrent=3)
|
|||
|
|
|
|||
|
|
processing_time = time.time() - start_time
|
|||
|
|
|
|||
|
|
print(f"✅ Batch processing completed in {processing_time:.2f}s")
|
|||
|
|
print(f"✅ Successfully processed {len([r for r in results if r])} out of {len(urls)} URLs")
|
|||
|
|
|
|||
|
|
for i, (url, result) in enumerate(zip(urls, results)):
|
|||
|
|
if result:
|
|||
|
|
script_status = "✅ JS executed" if result.script_result else "➖ No JS"
|
|||
|
|
word_count = result.word_count
|
|||
|
|
print(f" {i+1}. {url[:50]:<50} | {word_count:>4} words | {script_status}")
|
|||
|
|
if result.script_result:
|
|||
|
|
print(f" Script result: {str(result.script_result)[:80]}")
|
|||
|
|
else:
|
|||
|
|
print(f" {i+1}. {url[:50]:<50} | FAILED")
|
|||
|
|
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": "Batch Processing",
|
|||
|
|
"status": "success",
|
|||
|
|
"urls_processed": len([r for r in results if r]),
|
|||
|
|
"total_urls": len(urls),
|
|||
|
|
"processing_time": processing_time,
|
|||
|
|
"details": f"Mixed JS/no-JS processing successful"
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"❌ Batch processing failed: {e}")
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": "Batch Processing",
|
|||
|
|
"status": "failed",
|
|||
|
|
"error": str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
async def test_discovery_scenarios(self):
|
|||
|
|
"""Test discover() function with JavaScript enhancement."""
|
|||
|
|
print("\n🔍 Testing Discovery with JavaScript Enhancement")
|
|||
|
|
print("-" * 50)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
print("Testing discover() function (Note: May be limited implementation)")
|
|||
|
|
|
|||
|
|
# Test basic discovery
|
|||
|
|
start_time = time.time()
|
|||
|
|
results = await web.discover("Python web scraping", max_pages=3)
|
|||
|
|
discovery_time = time.time() - start_time
|
|||
|
|
|
|||
|
|
print(f"✅ Discovery completed in {discovery_time:.2f}s")
|
|||
|
|
print(f"✅ Found {len(results)} results")
|
|||
|
|
|
|||
|
|
for i, result in enumerate(results[:3]):
|
|||
|
|
print(f" {i+1}. {result.title[:60]}")
|
|||
|
|
print(f" URL: {result.url}")
|
|||
|
|
print(f" Words: {result.word_count}")
|
|||
|
|
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": "Discovery Function",
|
|||
|
|
"status": "success",
|
|||
|
|
"results_found": len(results),
|
|||
|
|
"discovery_time": discovery_time
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
except NotImplementedError:
|
|||
|
|
print("ℹ️ Discovery function not yet fully implemented (expected)")
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": "Discovery Function",
|
|||
|
|
"status": "not_implemented",
|
|||
|
|
"note": "Expected - discovery may need search engine integration"
|
|||
|
|
})
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"❌ Discovery test failed: {e}")
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": "Discovery Function",
|
|||
|
|
"status": "failed",
|
|||
|
|
"error": str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
async def _run_test_case(self, test: Dict[str, Any]):
|
|||
|
|
"""Run an individual test case."""
|
|||
|
|
print(f"\n🌐 Testing: {test['name']}")
|
|||
|
|
print(f" URL: {test['url']}")
|
|||
|
|
|
|||
|
|
start_time = time.time()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
if test['use_js'] and 'script' in test:
|
|||
|
|
print(f" JavaScript: {test['script'][:60]}...")
|
|||
|
|
content = await web.get(
|
|||
|
|
test['url'],
|
|||
|
|
script=test['script'],
|
|||
|
|
timeout=45
|
|||
|
|
)
|
|||
|
|
else:
|
|||
|
|
print(" Mode: Static content extraction")
|
|||
|
|
content = await web.get(test['url'], timeout=30)
|
|||
|
|
|
|||
|
|
load_time = time.time() - start_time
|
|||
|
|
|
|||
|
|
# Analyze results
|
|||
|
|
found_elements = sum(1 for element in test['expected_elements']
|
|||
|
|
if element.lower() in content.text.lower())
|
|||
|
|
|
|||
|
|
print(f" ✅ Loaded in {load_time:.2f}s")
|
|||
|
|
print(f" ✅ Title: {content.title}")
|
|||
|
|
print(f" ✅ Content: {content.word_count} words")
|
|||
|
|
print(f" ✅ Expected elements found: {found_elements}/{len(test['expected_elements'])}")
|
|||
|
|
|
|||
|
|
if content.script_result:
|
|||
|
|
print(f" ✅ JavaScript result: {str(content.script_result)[:100]}")
|
|||
|
|
|
|||
|
|
if content.script_error:
|
|||
|
|
print(f" ⚠️ JavaScript error: {content.script_error}")
|
|||
|
|
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": test['name'],
|
|||
|
|
"url": test['url'],
|
|||
|
|
"status": "success",
|
|||
|
|
"load_time": load_time,
|
|||
|
|
"word_count": content.word_count,
|
|||
|
|
"elements_found": found_elements,
|
|||
|
|
"expected_elements": len(test['expected_elements']),
|
|||
|
|
"has_js_result": content.script_result is not None,
|
|||
|
|
"has_js_error": content.script_error is not None
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
except Exception as e:
|
|||
|
|
load_time = time.time() - start_time
|
|||
|
|
print(f" ❌ Failed after {load_time:.2f}s: {e}")
|
|||
|
|
|
|||
|
|
self.results.append({
|
|||
|
|
"test_name": test['name'],
|
|||
|
|
"url": test['url'],
|
|||
|
|
"status": "failed",
|
|||
|
|
"load_time": load_time,
|
|||
|
|
"error": str(e)
|
|||
|
|
})
|
|||
|
|
|
|||
|
|
def print_summary(self):
|
|||
|
|
"""Print comprehensive test results summary."""
|
|||
|
|
print("\n" + "="*80)
|
|||
|
|
print("🎯 REAL-WORLD TESTING SUMMARY")
|
|||
|
|
print("="*80)
|
|||
|
|
|
|||
|
|
total_tests = len(self.results)
|
|||
|
|
successful_tests = len([r for r in self.results if r['status'] == 'success'])
|
|||
|
|
failed_tests = len([r for r in self.results if r['status'] == 'failed'])
|
|||
|
|
not_implemented = len([r for r in self.results if r['status'] == 'not_implemented'])
|
|||
|
|
|
|||
|
|
success_rate = (successful_tests / total_tests * 100) if total_tests > 0 else 0
|
|||
|
|
|
|||
|
|
print(f"\n📊 OVERALL RESULTS:")
|
|||
|
|
print(f" Total tests: {total_tests}")
|
|||
|
|
print(f" ✅ Successful: {successful_tests}")
|
|||
|
|
print(f" ❌ Failed: {failed_tests}")
|
|||
|
|
print(f" ℹ️ Not implemented: {not_implemented}")
|
|||
|
|
print(f" 📈 Success rate: {success_rate:.1f}%")
|
|||
|
|
|
|||
|
|
if successful_tests > 0:
|
|||
|
|
successful_results = [r for r in self.results if r['status'] == 'success']
|
|||
|
|
avg_load_time = sum(r.get('load_time', 0) for r in successful_results) / len(successful_results)
|
|||
|
|
total_words = sum(r.get('word_count', 0) for r in successful_results)
|
|||
|
|
js_enabled_tests = len([r for r in successful_results if r.get('has_js_result', False)])
|
|||
|
|
|
|||
|
|
print(f"\n⚡ PERFORMANCE METRICS:")
|
|||
|
|
print(f" Average load time: {avg_load_time:.2f}s")
|
|||
|
|
print(f" Total content extracted: {total_words:,} words")
|
|||
|
|
print(f" JavaScript-enhanced extractions: {js_enabled_tests}")
|
|||
|
|
|
|||
|
|
print(f"\n📋 DETAILED RESULTS:")
|
|||
|
|
for result in self.results:
|
|||
|
|
status_icon = "✅" if result['status'] == 'success' else "❌" if result['status'] == 'failed' else "ℹ️"
|
|||
|
|
print(f" {status_icon} {result['test_name']}")
|
|||
|
|
|
|||
|
|
if result['status'] == 'success':
|
|||
|
|
load_time = result.get('load_time', 0)
|
|||
|
|
words = result.get('word_count', 0)
|
|||
|
|
js_indicator = " (JS)" if result.get('has_js_result', False) else ""
|
|||
|
|
print(f" {load_time:.2f}s | {words} words{js_indicator}")
|
|||
|
|
elif result['status'] == 'failed':
|
|||
|
|
print(f" Error: {result.get('error', 'Unknown error')}")
|
|||
|
|
|
|||
|
|
print(f"\n🎉 JavaScript API Enhancement: {'VALIDATED' if success_rate >= 70 else 'NEEDS IMPROVEMENT'}")
|
|||
|
|
|
|||
|
|
if success_rate >= 70:
|
|||
|
|
print(" The JavaScript API enhancement is working well in real-world scenarios!")
|
|||
|
|
else:
|
|||
|
|
print(" Some issues detected that may need attention.")
|
|||
|
|
|
|||
|
|
async def main():
|
|||
|
|
"""Run comprehensive real-world testing."""
|
|||
|
|
print("🚀 Crawailer JavaScript API Enhancement - Real-World Testing")
|
|||
|
|
print("="*80)
|
|||
|
|
print(f"Test started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
|||
|
|
print("Testing enhanced JavaScript capabilities with real websites...")
|
|||
|
|
|
|||
|
|
tester = RealWorldTester()
|
|||
|
|
tester.test_start_time = time.time()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
# Run all test suites
|
|||
|
|
await tester.test_static_content_baseline()
|
|||
|
|
await tester.test_dynamic_content_scenarios()
|
|||
|
|
await tester.test_spa_and_modern_sites()
|
|||
|
|
await tester.test_batch_processing()
|
|||
|
|
await tester.test_discovery_scenarios()
|
|||
|
|
|
|||
|
|
except KeyboardInterrupt:
|
|||
|
|
print("\n⚠️ Testing interrupted by user")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"\n💥 Unexpected error during testing: {e}")
|
|||
|
|
import traceback
|
|||
|
|
traceback.print_exc()
|
|||
|
|
finally:
|
|||
|
|
total_time = time.time() - tester.test_start_time
|
|||
|
|
print(f"\nTotal testing time: {total_time:.2f}s")
|
|||
|
|
tester.print_summary()
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
print("Note: This requires Playwright to be installed and browser setup complete.")
|
|||
|
|
print("Run 'playwright install chromium' if you haven't already.")
|
|||
|
|
print()
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
asyncio.run(main())
|
|||
|
|
except KeyboardInterrupt:
|
|||
|
|
print("\nTesting cancelled by user.")
|
|||
|
|
except Exception as e:
|
|||
|
|
print(f"Failed to start testing: {e}")
|
|||
|
|
print("Make sure Playwright is properly installed and configured.")
|