test_browser_workflow_test.py•62.4 kB
#!/usr/bin/env python3
"""
End-to-End Test for Session-Based Browser Automation MCP Server
Tests the complete workflow from session creation to browser automation
"""
import unittest
import tempfile
from pathlib import Path
from datetime import datetime
import asyncio
import uuid
# Import the session-based browser manager
from browser_fastmcp_server import SessionBrowserManager, BrowserConfig
class MockContext:
"""Mock FastMCP Context for testing"""
def __init__(self, session_id: str):
self.session_id = session_id
self.logs = []
async def info(self, message: str):
self.logs.append(f"INFO: {message}")
print(f"INFO: {message}")
async def debug(self, message: str):
self.logs.append(f"DEBUG: {message}")
print(f"DEBUG: {message}")
class TestSessionBrowserWorkflow(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self):
self.manager = SessionBrowserManager(max_instances=5, default_ttl=300)
self.temp_dir = tempfile.mkdtemp(prefix="session_e2e_test_")
self.test_html_path = None
self.test_results = {}
await self.manager.start_cleanup_task()
await self.create_comprehensive_test_html()
async def asyncTearDown(self):
await self.manager.shutdown()
async def create_comprehensive_test_html(self):
"""Create comprehensive test HTML with various test elements"""
html_content = """
<!DOCTYPE html>
<html>
<head>
<title>Session Browser E2E Test Page</title>
<style>
body {
font-family: Arial, sans-serif;
margin: 20px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
min-height: 2000px;
}
.container {
max-width: 1200px;
margin: 0 auto;
background: white;
padding: 40px;
border-radius: 10px;
box-shadow: 0 10px 30px rgba(0,0,0,0.2);
}
.header {
background: linear-gradient(45deg, #ff6b6b, #4ecdc4);
color: white;
padding: 30px;
text-align: center;
border-radius: 8px;
margin-bottom: 30px;
}
.section {
margin: 30px 0;
padding: 20px;
border: 2px solid #e0e0e0;
border-radius: 8px;
background: #f9f9f9;
}
.screenshot-target {
background: linear-gradient(45deg, #74b9ff, #0984e3);
color: white;
padding: 20px;
border-radius: 8px;
margin: 15px 0;
text-align: center;
font-weight: bold;
}
.button-group {
display: flex;
gap: 15px;
flex-wrap: wrap;
margin: 20px 0;
}
.test-button {
background: #fd79a8;
color: white;
border: none;
padding: 12px 24px;
border-radius: 6px;
cursor: pointer;
font-size: 14px;
transition: all 0.3s;
}
.test-button:hover {
background: #e84393;
transform: translateY(-2px);
}
.form-section {
background: #dff9fb;
padding: 20px;
border-radius: 8px;
border: 2px solid #00b894;
}
.chart-container {
background: #2d3436;
color: white;
padding: 30px;
border-radius: 8px;
margin: 20px 0;
position: relative;
}
.chart-bar {
background: linear-gradient(45deg, #00cec9, #55a3ff);
height: 30px;
margin: 10px 0;
border-radius: 4px;
display: flex;
align-items: center;
padding-left: 15px;
color: white;
font-weight: bold;
}
.wide-content {
width: 1500px;
background: #ffeaa7;
padding: 20px;
margin: 20px 0;
border-radius: 8px;
overflow-x: auto;
}
.floating-element {
position: fixed;
top: 20px;
right: 20px;
background: #e17055;
color: white;
padding: 15px;
border-radius: 50%;
z-index: 1000;
width: 60px;
height: 60px;
display: flex;
align-items: center;
justify-content: center;
font-weight: bold;
}
.footer {
background: #636e72;
color: white;
padding: 40px;
text-align: center;
margin-top: 50px;
border-radius: 8px;
}
</style>
</head>
<body>
<div class="floating-element">📸</div>
<div class="container">
<div class="header" id="main-header">
<h1>Session Browser E2E Test Dashboard</h1>
<p>Comprehensive test page for session-based browser automation</p>
</div>
<div class="section" id="target-section">
<h2>🎯 Screenshot Target Tests</h2>
<div class="screenshot-target" id="target-1">
Target Element #1 - Primary Screenshot Target
</div>
<div class="screenshot-target" id="target-2">
Target Element #2 - Secondary Screenshot Target
</div>
<div class="screenshot-target" id="target-3">
Target Element #3 - Tertiary Screenshot Target
</div>
</div>
<div class="section" id="button-section">
<h2>🔘 Interactive Elements</h2>
<div class="button-group">
<button class="test-button" id="btn-1" onclick="console.log('Button 1 clicked')">Button 1</button>
<button class="test-button" id="btn-2" onclick="console.log('Button 2 clicked')">Button 2</button>
<button class="test-button" id="btn-3" onclick="console.log('Button 3 clicked')">Button 3</button>
<button class="test-button" id="btn-4" onclick="console.log('Button 4 clicked')">Button 4</button>
<button class="test-button" id="btn-5" onclick="console.log('Button 5 clicked')">Button 5</button>
</div>
</div>
<div class="section" id="form-section">
<h2>📝 Form Elements</h2>
<div class="form-section">
<input type="text" id="test-input" placeholder="Test input field" style="width: 100%; padding: 10px; margin: 10px 0;">
<select id="test-select" style="width: 100%; padding: 10px; margin: 10px 0;">
<option value="">Choose option...</option>
<option value="opt1">Option 1</option>
<option value="opt2">Option 2</option>
<option value="opt3">Option 3</option>
</select>
<textarea id="test-textarea" placeholder="Test textarea" style="width: 100%; height: 100px; padding: 10px; margin: 10px 0;"></textarea>
<input type="file" id="test-file" style="width: 100%; padding: 10px; margin: 10px 0;">
</div>
</div>
<div class="section" id="chart-section">
<h2>📊 Chart/Dashboard Simulation</h2>
<div class="chart-container">
<h3>Performance Metrics</h3>
<div class="chart-bar" style="width: 80%;">CPU Usage: 80%</div>
<div class="chart-bar" style="width: 65%;">Memory: 65%</div>
<div class="chart-bar" style="width: 90%;">Disk I/O: 90%</div>
<div class="chart-bar" style="width: 45%;">Network: 45%</div>
</div>
</div>
<div class="section" id="wide-section">
<h2>📐 Wide Content Test</h2>
<div class="wide-content">
This is a wide content area that extends beyond the normal viewport width to test horizontal scrolling and full-page screenshot capabilities. The content is intentionally wide to simulate dashboard layouts that require horizontal scrolling.
</div>
</div>
<div class="section" id="content-section">
<h2>📄 Content Extraction Test</h2>
<p>This section contains text content for extraction testing.</p>
<p id="scroll-target">Target text for scroll testing</p>
<ul>
<li>List item 1</li>
<li>List item 2</li>
<li>List item 3</li>
</ul>
<table border="1" style="width: 100%; border-collapse: collapse;">
<tr>
<th>Column 1</th>
<th>Column 2</th>
<th>Column 3</th>
</tr>
<tr>
<td>Data 1</td>
<td>Data 2</td>
<td>Data 3</td>
</tr>
<tr>
<td>Data 4</td>
<td>Data 5</td>
<td>Data 6</td>
</tr>
</table>
</div>
<div class="footer" id="footer-section">
<h3>🏁 End of Test Page</h3>
<p>Footer content for full-page screenshot testing</p>
</div>
</div>
<script>
// Add some dynamic content
document.addEventListener('DOMContentLoaded', function() {
const timestamp = new Date().toLocaleString();
document.querySelector('.header h1').innerHTML += `<br><small>Generated: ${timestamp}</small>`;
});
</script>
</body>
</html>
"""
html_file = Path(self.temp_dir) / "comprehensive_test.html"
html_file.write_text(html_content)
self.test_html_path = f"file://{html_file.absolute()}"
print(f"📄 Created comprehensive test HTML: {self.test_html_path}")
async def test_complete_workflow(self):
"""Test complete browser automation workflow"""
print("\n🧪 Testing complete browser automation workflow...")
session_id = f"e2e_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Step 1: Start browser session
print("\n📱 Step 1: Starting browser session...")
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
await ctx.info(f"Browser session started for {session_id}")
# Step 2: Navigate to a test page
print("\n🌐 Step 2: Navigating to test page...")
browser_session = instance.browser_session
await browser_session.navigate("https://httpbin.org/html")
await ctx.info("Navigated to test page")
# Wait for page to load
await asyncio.sleep(2)
# Step 3: Get page state
print("\n📄 Step 3: Getting page state...")
try:
state_summary = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
await ctx.info(f"Interactive elements: {len(state_summary.selector_map)}")
except Exception as e:
await ctx.info(f"Could not get state summary: {e}")
# Step 4: Take screenshot
print("\n📸 Step 4: Taking screenshot...")
try:
page = await browser_session.get_current_page()
screenshot_bytes = await page.screenshot(full_page=True)
screenshot_path = Path(instance.temp_dir) / "e2e_test_screenshot.png"
screenshot_path.write_bytes(screenshot_bytes)
await ctx.info(f"Screenshot saved: {screenshot_path}")
except Exception as e:
await ctx.info(f"Screenshot failed: {e}")
# Step 5: Extract content
print("\n📝 Step 5: Extracting content...")
try:
page = await browser_session.get_current_page()
html_content = await page.content()
title = await page.title()
await ctx.info(f"Page title: {title}")
await ctx.info(f"Content length: {len(html_content)} characters")
except Exception as e:
await ctx.info(f"Content extraction failed: {e}")
# Step 6: Test cookie management
print("\n🍪 Step 6: Testing cookie management...")
try:
page = await browser_session.get_current_page()
context = page.context
test_cookie = {
"name": "test_cookie",
"value": "test_value",
"domain": "httpbin.org",
"path": "/"
}
await context.add_cookies([test_cookie])
await ctx.info("Test cookie set")
cookies = await context.cookies()
await ctx.info(f"Found {len(cookies)} cookies")
except Exception as e:
await ctx.info(f"Cookie management failed: {e}")
# Step 7: Test tab management
print("\n📑 Step 7: Testing tab management...")
try:
new_page = await browser_session.navigate("https://httpbin.org/json")
await ctx.info("New tab created")
tabs = browser_session.tabs
await ctx.info(f"Total tabs: {len(tabs)}")
except Exception as e:
await ctx.info(f"Tab management failed: {e}")
# Step 8: Test file operations
print("\n📁 Step 8: Testing file operations...")
test_file = Path(instance.temp_dir) / "test_download.txt"
test_file.write_text("This is a test download file")
await ctx.info(f"Test file created: {test_file}")
# Step 9: Test PDF generation
print("\n📄 Step 9: Testing PDF generation...")
try:
page = await browser_session.get_current_page()
pdf_bytes = await page.pdf()
pdf_path = Path(instance.temp_dir) / "e2e_test.pdf"
pdf_path.write_bytes(pdf_bytes)
await ctx.info(f"PDF generated: {pdf_path}")
except Exception as e:
await ctx.info(f"PDF generation failed: {e}")
# Step 10: Cleanup
print("\n🧹 Step 10: Cleaning up...")
await self.manager.close_session(session_id)
await ctx.info("Session cleaned up")
try:
info = await self.manager.get_session_info(session_id)
assert info is None, "Session should be cleaned up"
except Exception as e:
if "not found" in str(e):
await ctx.info("Session properly cleaned up")
else:
raise e
await ctx.info("✅ Complete workflow test passed!")
async def test_multiple_sessions(self):
"""Test multiple concurrent sessions"""
print("\n🧪 Testing multiple concurrent sessions...")
sessions = ["session_1", "session_2", "session_3"]
instances = []
# Create multiple sessions
for session_id in sessions:
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
instances.append(instance)
print(f"✅ Created session: {session_id}")
# Verify isolation
for i, instance in enumerate(instances):
assert instance.session_id == sessions[i]
assert instance.temp_dir != instances[(i+1)%len(instances)].temp_dir
print("✅ All sessions are properly isolated")
# Test concurrent operations
tasks = []
for i, instance in enumerate(instances):
task = self._test_session_operations(instance, f"Session {i+1}")
tasks.append(task)
results = await asyncio.gather(*tasks, return_exceptions=True)
# Check results
success_count = 0
for i, result in enumerate(results):
if isinstance(result, Exception):
print(f"❌ Session {i+1} failed: {result}")
else:
print(f"✅ Session {i+1} operations completed")
success_count += 1
# Cleanup
for session_id in sessions:
await self.manager.close_session(session_id)
# Consider test passed if at least 2 sessions worked
success = success_count >= 2
self.assertEqual(success_count, 3)
async def _test_session_operations(self, instance, session_name: str):
"""Test basic operations for a session"""
try:
browser_session = instance.browser_session
# Navigate to a simple page
await browser_session.navigate("https://httpbin.org/get")
await asyncio.sleep(1)
# Get page info
page = await browser_session.get_current_page()
title = await page.title()
# Take screenshot
screenshot_bytes = await page.screenshot()
screenshot_path = Path(instance.temp_dir) / f"{session_name.lower().replace(' ', '_')}_screenshot.png"
screenshot_path.write_bytes(screenshot_bytes)
return True
except Exception as e:
print(f"❌ {session_name} operations failed: {e}")
return False
async def test_session_isolation(self):
"""Test that sessions are properly isolated"""
print("\n🧪 Testing session isolation...")
# Create two sessions
session1 = f"isolation_test_{uuid.uuid4()}"
session2 = f"isolation_test_{uuid.uuid4()}"
instance1 = await self.manager.get_or_create_session_instance(session1)
instance2 = await self.manager.get_or_create_session_instance(session2)
# Verify different temp directories
assert instance1.temp_dir != instance2.temp_dir, "Sessions should have different temp directories"
# Verify different browser sessions
assert instance1.browser_session is not instance2.browser_session, "Sessions should have different browser sessions"
# Test that they can navigate to different pages independently
print(" Navigating session 1 to html page...")
await instance1.browser_session.navigate("https://httpbin.org/html")
await asyncio.sleep(3) # Wait longer for navigation
print(" Navigating session 2 to json page...")
await instance2.browser_session.navigate("https://httpbin.org/json")
await asyncio.sleep(3) # Wait longer for navigation
# Get pages and check URLs
page1 = await instance1.browser_session.get_current_page()
page2 = await instance2.browser_session.get_current_page()
url1 = page1.url
url2 = page2.url
print(f" Session 1 URL: {url1}")
print(f" Session 2 URL: {url2}")
# More flexible URL checking - just verify they're on different pages
assert "httpbin.org" in url1, f"Session 1 should be on httpbin.org, got: {url1}"
# For session 2, be more lenient - it might still be loading
if "about:blank" in url2:
print(" Session 2 still loading, waiting a bit more...")
await asyncio.sleep(2)
page2 = await instance2.browser_session.get_current_page()
url2 = page2.url
print(f" Session 2 URL after wait: {url2}")
# Final check - both should be on httpbin.org but different pages
assert "httpbin.org" in url1, f"Session 1 should be on httpbin.org, got: {url1}"
assert "httpbin.org" in url2, f"Session 2 should be on httpbin.org, got: {url2}"
# Check that URLs are different (they should be on different pages)
assert url1 != url2, f"Session URLs should be different, both got: {url1}"
print("✅ Session isolation verified - sessions are on different pages")
# Cleanup
await self.manager.close_session(session1)
await self.manager.close_session(session2)
async def test_comprehensive_screenshot_functionality(self):
"""Test comprehensive screenshot functionality"""
print("\n🧪 Testing comprehensive screenshot functionality...")
session_id = f"screenshot_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
browser_session = instance.browser_session
await browser_session.navigate(self.test_html_path)
await asyncio.sleep(3)
page = await browser_session.get_current_page()
screenshot_tests = []
# Test 1: Full page screenshot
try:
screenshot_bytes = await page.screenshot(full_page=True, type="png")
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"e2e_fullpage_{timestamp}.png"
file_path = Path(instance.temp_dir) / filename
file_path.write_bytes(screenshot_bytes)
screenshot_tests.append(f"Full page: {filename} ({len(screenshot_bytes)/1024:.1f} KB)")
except Exception as e:
screenshot_tests.append(f"Full page: FAILED - {e}")
# Test 2: Viewport screenshot
try:
await page.set_viewport_size({"width": 1024, "height": 768})
screenshot_bytes = await page.screenshot(full_page=False, type="png")
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"e2e_viewport_{timestamp}.png"
file_path = Path(instance.temp_dir) / filename
file_path.write_bytes(screenshot_bytes)
screenshot_tests.append(f"Viewport: {filename} ({len(screenshot_bytes)/1024:.1f} KB)")
except Exception as e:
screenshot_tests.append(f"Viewport: FAILED - {e}")
# Test 3: Target element screenshot
try:
await page.wait_for_selector("#target-1", timeout=5000)
element = page.locator("#target-1").first
screenshot_bytes = await element.screenshot(type="png")
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"e2e_target_{timestamp}.png"
file_path = Path(instance.temp_dir) / filename
file_path.write_bytes(screenshot_bytes)
screenshot_tests.append(f"Target element: {filename} ({len(screenshot_bytes)/1024:.1f} KB)")
except Exception as e:
screenshot_tests.append(f"Target element: FAILED - {e}")
# Test 4: JPEG screenshot
try:
screenshot_bytes = await page.screenshot(full_page=False, type="jpeg", quality=85)
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"e2e_jpeg_{timestamp}.jpeg"
file_path = Path(instance.temp_dir) / filename
file_path.write_bytes(screenshot_bytes)
screenshot_tests.append(f"JPEG: {filename} ({len(screenshot_bytes)/1024:.1f} KB)")
except Exception as e:
screenshot_tests.append(f"JPEG: FAILED - {e}")
# Test 5: Multiple target screenshots
try:
targets = ["#target-1", "#target-2", "#target-3"]
screenshots_taken = 0
for i, target in enumerate(targets):
try:
await page.wait_for_selector(target, timeout=3000)
element = page.locator(target).first
screenshot_bytes = await element.screenshot(type="png")
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"e2e_multi_{i+1}_{timestamp}.png"
file_path = Path(instance.temp_dir) / filename
file_path.write_bytes(screenshot_bytes)
screenshots_taken += 1
except Exception:
continue
screenshot_tests.append(f"Multiple targets: {screenshots_taken}/{len(targets)} successful")
except Exception as e:
screenshot_tests.append(f"Multiple targets: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in screenshot_tests if "FAILED" not in test)
self.assertEqual(success_count, 5)
async def test_browser_actions_coverage(self):
"""Test comprehensive browser actions coverage"""
print("\n🧪 Testing browser actions coverage...")
session_id = f"browser_actions_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
browser_session = instance.browser_session
await browser_session.navigate(self.test_html_path)
await asyncio.sleep(3)
page = await browser_session.get_current_page()
action_tests = []
# Test 1: Page state extraction
try:
state = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
elements_count = len(state.selector_map)
action_tests.append(f"Page state: {elements_count} interactive elements found")
except Exception as e:
action_tests.append(f"Page state: FAILED - {e}")
# Test 2: Element clicking
try:
state = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
if state.selector_map:
first_idx = list(state.selector_map.keys())[0]
element = await browser_session.get_dom_element_by_index(first_idx)
if element:
element_text = element.get_all_text_till_next_clickable_element()[:30]
action_tests.append(f"Element clicking: accessible element '{element_text}'")
else:
action_tests.append("Element clicking: element not accessible")
else:
action_tests.append("Element clicking: no clickable elements found")
except Exception as e:
action_tests.append(f"Element clicking: FAILED - {e}")
# Test 3: Text input
try:
state = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
input_idx = None
for idx, element in state.selector_map.items():
if element.tag_name.lower() in ['input', 'textarea']:
input_idx = idx
break
if input_idx is not None:
element = await browser_session.get_dom_element_by_index(input_idx)
if element:
await browser_session._input_text_element_node(element, "E2E Test Input")
action_tests.append(f"Text input: successful to element {input_idx}")
else:
action_tests.append("Text input: element not accessible")
else:
action_tests.append("Text input: no input elements found")
except Exception as e:
action_tests.append(f"Text input: FAILED - {e}")
# Test 4: Scrolling
try:
initial_scroll = await page.evaluate('() => window.pageYOffset')
dy = await page.evaluate('() => window.innerHeight')
await page.evaluate('(y) => window.scrollBy(0, y)', dy)
await asyncio.sleep(0.5)
new_scroll = await page.evaluate('() => window.pageYOffset')
if new_scroll > initial_scroll:
action_tests.append(f"Scrolling: successful ({initial_scroll} → {new_scroll})")
else:
action_tests.append("Scrolling: position unchanged")
except Exception as e:
action_tests.append(f"Scrolling: FAILED - {e}")
# Test 5: Keyboard input
try:
test_keys = ["Tab", "Enter", "Escape"]
results = []
for key in test_keys:
try:
await page.keyboard.press(key)
results.append(key)
except Exception:
continue
action_tests.append(f"Keyboard input: {len(results)}/{len(test_keys)} keys successful")
except Exception as e:
action_tests.append(f"Keyboard input: FAILED - {e}")
# Test 6: Content extraction
try:
html_content = await page.content()
try:
import markdownify
markdown = markdownify.markdownify(html_content, strip=['script', 'style'])
action_tests.append(f"Content extraction: {len(markdown)} characters extracted")
except ImportError:
action_tests.append(f"Content extraction: {len(html_content)} characters extracted (no markdown)")
except Exception as e:
action_tests.append(f"Content extraction: FAILED - {e}")
# Test 7: Scroll to text
try:
locator = page.get_by_text("Target text", exact=False)
count = await locator.count()
action_tests.append(f"Scroll to text: {count} text matches found")
except Exception as e:
action_tests.append(f"Scroll to text: FAILED - {e}")
# Test 8: Dropdown detection
try:
state = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
select_found = False
for idx, element in state.selector_map.items():
if element.tag_name.lower() == 'select':
select_found = True
try:
options = await page.evaluate(
f"""
() => {{
const select = document.evaluate('{element.xpath}', document, null,
XPathResult.FIRST_ORDERED_NODE_TYPE, null).singleNodeValue;
if (!select) return null;
return Array.from(select.options).map(opt => opt.text);
}}
"""
)
if options:
action_tests.append(f"Dropdown detection: {len(options)} options found")
else:
action_tests.append("Dropdown detection: no options found")
except Exception:
action_tests.append("Dropdown detection: options extraction failed")
break
if not select_found:
action_tests.append("Dropdown detection: no select elements found")
except Exception as e:
action_tests.append(f"Dropdown detection: FAILED - {e}")
# Test 9: File upload detection
try:
state = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
file_input_found = False
for idx, element in state.selector_map.items():
if element.tag_name.lower() == 'input' and 'file' in str(element).lower():
file_input_found = True
break
if file_input_found:
action_tests.append("File upload: file input element detected")
else:
action_tests.append("File upload: no file input found")
except Exception as e:
action_tests.append(f"File upload: FAILED - {e}")
# Test 10: Tab management
try:
initial_tabs = len(browser_session.tabs)
await browser_session.create_new_tab("about:blank")
new_tab_count = len(browser_session.tabs)
action_tests.append(f"Tab management: {initial_tabs} → {new_tab_count} tabs")
except Exception as e:
action_tests.append(f"Tab management: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in action_tests if "FAILED" not in test)
self.assertEqual(success_count, 10)
async def test_browser_configuration(self):
"""Test browser configuration functionality"""
print("\n🧪 Testing browser configuration...")
session_id = f"config_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
config_tests = []
# Test 1: Default configuration
try:
default_config = BrowserConfig()
instance = await self.manager.get_or_create_session_instance(
session_id + "_default", default_config
)
config_tests.append(f"Default config: headless={default_config.headless}, viewport={default_config.viewport_width}x{default_config.viewport_height}")
await self.manager.close_session(session_id + "_default")
except Exception as e:
config_tests.append(f"Default config: FAILED - {e}")
# Test 2: Custom configuration
try:
custom_config = BrowserConfig(
headless=True,
viewport_width=1280,
viewport_height=720,
disable_gpu=True,
user_agent="E2E-Test-Agent/1.0"
)
instance = await self.manager.get_or_create_session_instance(
session_id + "_custom", custom_config
)
# Verify configuration was applied
page = await instance.browser_session.get_current_page()
viewport = page.viewport_size
user_agent = await page.evaluate("() => navigator.userAgent")
config_tests.append(f"Custom config: viewport={viewport}, user_agent contains 'E2E-Test': {'E2E-Test' in user_agent}")
await self.manager.close_session(session_id + "_custom")
except Exception as e:
config_tests.append(f"Custom config: FAILED - {e}")
# Test 3: Security configurations
try:
security_config = BrowserConfig(
headless=True,
disable_web_security=True,
no_sandbox=True,
disable_dev_shm_usage=True
)
instance = await self.manager.get_or_create_session_instance(
session_id + "_security", security_config
)
config_tests.append("Security config: applied successfully")
await self.manager.close_session(session_id + "_security")
except Exception as e:
config_tests.append(f"Security config: FAILED - {e}")
# Test 4: Performance configurations
try:
perf_config = BrowserConfig(
headless=True,
disable_gpu=True,
disable_extensions=True,
disable_plugins=True,
disable_background_timer_throttling=True
)
instance = await self.manager.get_or_create_session_instance(
session_id + "_perf", perf_config
)
config_tests.append("Performance config: applied successfully")
await self.manager.close_session(session_id + "_perf")
except Exception as e:
config_tests.append(f"Performance config: FAILED - {e}")
success_count = sum(1 for test in config_tests if "FAILED" not in test)
self.assertEqual(success_count, 4)
async def test_cookie_management(self):
"""Test comprehensive cookie management functionality"""
print("\n🧪 Testing cookie management...")
session_id = f"cookie_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
browser_session = instance.browser_session
await browser_session.navigate("https://httpbin.org/cookies/set/test/value")
await asyncio.sleep(2)
page = await browser_session.get_current_page()
context = page.context
cookie_tests = []
# Test 1: Set basic cookie
try:
test_cookie = {
"name": "e2e_test_cookie",
"value": "test_value_123",
"domain": "httpbin.org",
"path": "/"
}
await context.add_cookies([test_cookie])
cookie_tests.append("Basic cookie: set successfully")
except Exception as e:
cookie_tests.append(f"Basic cookie: FAILED - {e}")
# Test 2: Set secure cookie
try:
secure_cookie = {
"name": "secure_cookie",
"value": "secure_value",
"domain": "httpbin.org",
"path": "/",
"secure": True,
"httpOnly": True,
"sameSite": "Strict"
}
await context.add_cookies([secure_cookie])
cookie_tests.append("Secure cookie: set successfully")
except Exception as e:
cookie_tests.append(f"Secure cookie: FAILED - {e}")
# Test 3: Get all cookies
try:
cookies = await context.cookies()
cookie_count = len(cookies)
cookie_tests.append(f"Get cookies: found {cookie_count} cookies")
except Exception as e:
cookie_tests.append(f"Get cookies: FAILED - {e}")
# Test 4: Verify specific cookie
try:
cookies = await context.cookies()
test_cookie_found = any(c.get('name') == 'e2e_test_cookie' for c in cookies)
cookie_tests.append(f"Verify cookie: test cookie found = {test_cookie_found}")
except Exception as e:
cookie_tests.append(f"Verify cookie: FAILED - {e}")
# Test 5: Cookie persistence across navigation
try:
await browser_session.navigate("https://httpbin.org/cookies")
await asyncio.sleep(2)
cookies_after_nav = await context.cookies()
persistent_count = len(cookies_after_nav)
cookie_tests.append(f"Cookie persistence: {persistent_count} cookies after navigation")
except Exception as e:
cookie_tests.append(f"Cookie persistence: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in cookie_tests if "FAILED" not in test)
self.assertEqual(success_count, 5)
async def test_file_operations(self):
"""Test file upload and download operations"""
print("\n🧪 Testing file operations...")
session_id = f"file_ops_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
browser_session = instance.browser_session
file_tests = []
# Test 1: Create test file for upload
try:
test_file_path = Path(instance.temp_dir) / "test_upload.txt"
test_file_content = "This is a test file for upload testing\nGenerated at: " + datetime.now().isoformat()
test_file_path.write_text(test_file_content)
file_tests.append(f"Test file creation: {test_file_path.name} ({len(test_file_content)} bytes)")
except Exception as e:
file_tests.append(f"Test file creation: FAILED - {e}")
# Test 2: Navigate to file upload test page
try:
await browser_session.navigate(self.test_html_path)
await asyncio.sleep(2)
page = await browser_session.get_current_page()
file_tests.append("Navigation to test page: successful")
except Exception as e:
file_tests.append(f"Navigation to test page: FAILED - {e}")
# Test 3: File input detection
try:
state = await browser_session.get_state_summary(cache_clickable_elements_hashes=True)
file_inputs = []
for idx, element in state.selector_map.items():
if element.tag_name.lower() == 'input' and 'file' in str(element).lower():
file_inputs.append(idx)
file_tests.append(f"File input detection: found {len(file_inputs)} file input(s)")
except Exception as e:
file_tests.append(f"File input detection: FAILED - {e}")
# Test 4: File upload simulation (without actual upload)
try:
if test_file_path.exists():
# Just verify the file can be accessed for upload
file_size = test_file_path.stat().st_size
file_tests.append(f"File upload simulation: test file ready ({file_size} bytes)")
else:
file_tests.append("File upload simulation: test file not found")
except Exception as e:
file_tests.append(f"File upload simulation: FAILED - {e}")
# Test 5: Download simulation - create a downloadable file
try:
download_file_path = Path(instance.temp_dir) / "test_download.json"
download_content = {
"test": "download",
"timestamp": datetime.now().isoformat(),
"session_id": session_id
}
import json
download_file_path.write_text(json.dumps(download_content, indent=2))
download_size = download_file_path.stat().st_size
file_tests.append(f"Download simulation: created downloadable file ({download_size} bytes)")
except Exception as e:
file_tests.append(f"Download simulation: FAILED - {e}")
# Test 6: File type detection
try:
import mimetypes
test_mime = mimetypes.guess_type(str(test_file_path))[0]
download_mime = mimetypes.guess_type(str(download_file_path))[0]
file_tests.append(f"MIME type detection: txt={test_mime}, json={download_mime}")
except Exception as e:
file_tests.append(f"MIME type detection: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in file_tests if "FAILED" not in test)
self.assertEqual(success_count, 6)
async def test_pdf_generation(self):
"""Test PDF generation functionality"""
print("\n🧪 Testing PDF generation...")
session_id = f"pdf_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True) # Use headless for PDF generation
)
browser_session = instance.browser_session
pdf_tests = []
# Test 1: PDF from current page
try:
await browser_session.navigate(self.test_html_path)
await asyncio.sleep(3)
page = await browser_session.get_current_page()
pdf_bytes = await page.pdf()
pdf_path = Path(instance.temp_dir) / "test_current_page.pdf"
pdf_path.write_bytes(pdf_bytes)
pdf_size_kb = len(pdf_bytes) / 1024
pdf_tests.append(f"Current page PDF: {pdf_path.name} ({pdf_size_kb:.1f} KB)")
except Exception as e:
pdf_tests.append(f"Current page PDF: FAILED - {e}")
# Test 2: PDF with custom options
try:
pdf_options = {
"format": "A4",
"print_background": True,
"margin": {
"top": "1in",
"bottom": "1in",
"left": "0.5in",
"right": "0.5in"
}
}
pdf_bytes = await page.pdf(**pdf_options)
pdf_path = Path(instance.temp_dir) / "test_custom_options.pdf"
pdf_path.write_bytes(pdf_bytes)
pdf_size_kb = len(pdf_bytes) / 1024
pdf_tests.append(f"Custom options PDF: {pdf_path.name} ({pdf_size_kb:.1f} KB)")
except Exception as e:
pdf_tests.append(f"Custom options PDF: FAILED - {e}")
# Test 3: PDF from URL
try:
await browser_session.navigate("https://httpbin.org/html")
await asyncio.sleep(2)
page = await browser_session.get_current_page()
pdf_bytes = await page.pdf()
pdf_path = Path(instance.temp_dir) / "test_from_url.pdf"
pdf_path.write_bytes(pdf_bytes)
pdf_size_kb = len(pdf_bytes) / 1024
pdf_tests.append(f"URL PDF: {pdf_path.name} ({pdf_size_kb:.1f} KB)")
except Exception as e:
pdf_tests.append(f"URL PDF: FAILED - {e}")
# Test 4: PDF from HTML content
try:
html_content = """
<html>
<head><title>E2E Test PDF</title></head>
<body>
<h1>PDF Generation Test</h1>
<p>This PDF was generated during E2E testing.</p>
<p>Timestamp: """ + datetime.now().isoformat() + """</p>
</body>
</html>
"""
await page.set_content(html_content)
await page.wait_for_load_state(state='domcontentloaded')
pdf_bytes = await page.pdf()
pdf_path = Path(instance.temp_dir) / "test_from_html.pdf"
pdf_path.write_bytes(pdf_bytes)
pdf_size_kb = len(pdf_bytes) / 1024
pdf_tests.append(f"HTML content PDF: {pdf_path.name} ({pdf_size_kb:.1f} KB)")
except Exception as e:
pdf_tests.append(f"HTML content PDF: FAILED - {e}")
# Test 5: PDF metadata verification
try:
# Verify PDFs are valid by checking file headers
pdf_files = list(Path(instance.temp_dir).glob("*.pdf"))
valid_pdfs = 0
for pdf_file in pdf_files:
with open(pdf_file, 'rb') as f:
header = f.read(5)
if header == b'%PDF-':
valid_pdfs += 1
pdf_tests.append(f"PDF validation: {valid_pdfs}/{len(pdf_files)} files are valid PDFs")
except Exception as e:
pdf_tests.append(f"PDF validation: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in pdf_tests if "FAILED" not in test)
self.assertEqual(success_count, 5)
async def test_advanced_navigation(self):
"""Test advanced navigation functionality"""
print("\n🧪 Testing advanced navigation...")
session_id = f"nav_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
browser_session = instance.browser_session
nav_tests = []
# Test 1: Basic navigation
try:
await browser_session.navigate("https://httpbin.org/html")
await asyncio.sleep(2)
page = await browser_session.get_current_page()
nav_tests.append(f"Basic navigation: navigated to {page.url}")
except Exception as e:
nav_tests.append(f"Basic navigation: FAILED - {e}")
# Test 2: Navigate to another page for history
try:
await browser_session.navigate("https://httpbin.org/json")
await asyncio.sleep(2)
page = await browser_session.get_current_page()
nav_tests.append(f"Second navigation: navigated to {page.url}")
except Exception as e:
nav_tests.append(f"Second navigation: FAILED - {e}")
# Test 3: Navigate back
try:
await browser_session.go_back()
await asyncio.sleep(2)
page = await browser_session.get_current_page()
nav_tests.append(f"Navigate back: returned to {page.url}")
except Exception as e:
nav_tests.append(f"Navigate back: FAILED - {e}")
# Test 4: Navigate forward
try:
await page.go_forward()
await asyncio.sleep(2)
page = await browser_session.get_current_page()
nav_tests.append(f"Navigate forward: went to {page.url}")
except Exception as e:
nav_tests.append(f"Navigate forward: FAILED - {e}")
# Test 5: New tab navigation
try:
initial_tab_count = len(browser_session.tabs)
await browser_session.navigate("https://httpbin.org/user-agent")
new_page = await browser_session.get_current_page()
await asyncio.sleep(2)
new_tab_count = len(browser_session.tabs)
nav_tests.append(f"New tab navigation: {initial_tab_count} → {new_tab_count} tabs, URL: {new_page.url}")
except Exception as e:
nav_tests.append(f"New tab navigation: FAILED - {e}")
# Test 6: Tab switching
try:
if len(browser_session.tabs) > 1:
await browser_session.switch_to_tab(0)
await asyncio.sleep(1)
current_page = await browser_session.get_current_page()
nav_tests.append(f"Tab switching: switched to tab 0, URL: {current_page.url}")
else:
nav_tests.append("Tab switching: only one tab available")
except Exception as e:
nav_tests.append(f"Tab switching: FAILED - {e}")
# Test 7: Page refresh
try:
page = await browser_session.get_current_page()
await page.reload()
await asyncio.sleep(2)
nav_tests.append("Page refresh: successful")
except Exception as e:
nav_tests.append(f"Page refresh: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in nav_tests if "FAILED" not in test)
self.assertEqual(success_count, 7)
async def test_session_management_tools(self):
"""Test session management tools functionality"""
print("\n🧪 Testing session management tools...")
session_id = f"mgmt_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
mgmt_tests = []
# Test 1: Session creation and info
try:
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
session_info = await self.manager.get_session_info(session_id)
if session_info:
mgmt_tests.append(f"Session info: created at {session_info.created_at}, config headless={session_info.config.headless}")
else:
mgmt_tests.append("Session info: FAILED - no info found")
except Exception as e:
mgmt_tests.append(f"Session info: FAILED - {e}")
# Test 2: Multiple session creation
try:
session_ids = ["mgmt_test_1", "mgmt_test_2", "mgmt_test_3"]
created_sessions = []
for sid in session_ids:
instance = await self.manager.get_or_create_session_instance(sid)
created_sessions.append(sid)
mgmt_tests.append(f"Multiple sessions: created {len(created_sessions)} sessions")
except Exception as e:
mgmt_tests.append(f"Multiple sessions: FAILED - {e}")
# Test 3: Session count and status
try:
session_count = self.manager.get_session_count()
all_sessions = await self.manager.get_all_sessions_info()
mgmt_tests.append(f"Session status: {session_count} active sessions, {len(all_sessions)} info objects")
except Exception as e:
mgmt_tests.append(f"Session status: FAILED - {e}")
# Test 4: Session isolation verification
try:
if len(session_ids) >= 2:
info1 = await self.manager.get_session_info(session_ids[0])
info2 = await self.manager.get_session_info(session_ids[1])
if info1 and info2:
isolation_check = info1.temp_dir != info2.temp_dir
mgmt_tests.append(f"Session isolation: temp directories isolated = {isolation_check}")
else:
mgmt_tests.append("Session isolation: FAILED - missing session info")
else:
mgmt_tests.append("Session isolation: insufficient sessions for test")
except Exception as e:
mgmt_tests.append(f"Session isolation: FAILED - {e}")
# Test 5: Session cleanup
try:
cleanup_count = 0
for sid in session_ids:
await self.manager.close_session(sid)
cleanup_count += 1
# Verify cleanup
remaining_count = self.manager.get_session_count()
mgmt_tests.append(f"Session cleanup: cleaned {cleanup_count} sessions, {remaining_count} remaining")
except Exception as e:
mgmt_tests.append(f"Session cleanup: FAILED - {e}")
# Test 6: Session recreation after cleanup
try:
# Try to recreate a session that was just closed
new_instance = await self.manager.get_or_create_session_instance(session_ids[0])
new_info = await self.manager.get_session_info(session_ids[0])
if new_info:
mgmt_tests.append(f"Session recreation: successfully recreated session {session_ids[0]}")
else:
mgmt_tests.append("Session recreation: FAILED - no info after recreation")
await self.manager.close_session(session_ids[0])
except Exception as e:
mgmt_tests.append(f"Session recreation: FAILED - {e}")
# Final cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in mgmt_tests if "FAILED" not in test)
self.assertEqual(success_count, 6)
async def test_search_functionality(self):
"""Test search functionality"""
print("\n🧪 Testing search functionality...")
session_id = f"search_test_session_{uuid.uuid4()}"
ctx = MockContext(session_id)
# Start session
instance = await self.manager.get_or_create_session_instance(
session_id,
BrowserConfig(headless=True)
)
browser_session = instance.browser_session
search_tests = []
# Test 1: Google search navigation
try:
await browser_session.navigate("https://www.bing.com")
await asyncio.sleep(2)
page = await browser_session.get_current_page()
search_tests.append(f"Google navigation: navigated to {page.url}")
except Exception as e:
raise e
search_tests.append(f"Google navigation: FAILED - {e}")
# Test 2: Search query construction
try:
search_query = "browser automation testing"
search_url = f"https://www.bing.com/search?q={search_query}&udm=14"
await browser_session.navigate(search_url)
await asyncio.sleep(3)
page = await browser_session.get_current_page()
search_tests.append(f"Search query: constructed and navigated to search results")
except Exception as e:
raise e
search_tests.append(f"Search query: FAILED - {e}")
# Test 3: Search results page analysis
try:
page = await browser_session.get_current_page()
await page.wait_for_load_state('load') # 等待页面完全加载
title = await page.title()
html_content = await page.content()
print("title:", title)
print("html_content:", html_content)
# Check if we're on a search results page
is_search_page = "search" in title.lower() or "search" in html_content.lower()
search_tests.append(f"Search results analysis: is search page = {is_search_page}, title: {title[:50]}...")
except Exception as e:
raise e
search_tests.append(f"Search results analysis: FAILED - {e}")
# Test 4: Content extraction from search page
try:
page = await browser_session.get_current_page()
# Look for search-related elements
search_elements = await page.evaluate("""
() => {
const results = document.querySelectorAll('[data-result], .result, .search-result');
const links = document.querySelectorAll('a[href*="http"]');
return {
resultElements: results.length,
totalLinks: links.length
};
}
""")
search_tests.append(f"Content extraction: found {search_elements.get('resultElements', 0)} result elements, {search_elements.get('totalLinks', 0)} links")
except Exception as e:
raise e
search_tests.append(f"Content extraction: FAILED - {e}")
# Test 5: Search with different query
try:
alternative_query = "selenium webdriver"
alt_search_url = f"https://www.google.com/search?q={alternative_query}"
await browser_session.navigate(alt_search_url)
await asyncio.sleep(2)
page = await browser_session.get_current_page()
search_tests.append(f"Alternative search: performed search for '{alternative_query}'")
except Exception as e:
raise e
search_tests.append(f"Alternative search: FAILED - {e}")
# Test 6: Local content search simulation
try:
# Navigate to our test page and search within it
await browser_session.navigate(self.test_html_path)
await asyncio.sleep(2)
page = await browser_session.get_current_page()
# Search for specific text in the page
search_text = "Target text"
text_found = await page.evaluate(f"""
() => {{
const content = document.body.textContent || document.body.innerText;
return content.includes('{search_text}');
}}
""")
search_tests.append(f"Local content search: found '{search_text}' = {text_found}")
except Exception as e:
raise e
search_tests.append(f"Local content search: FAILED - {e}")
# Cleanup
await self.manager.close_session(session_id)
success_count = sum(1 for test in search_tests if "FAILED" not in test)
self.assertEqual(success_count, 6)