refactor(browser): consolidate browser strategy implementations

Moves common browser functionality into BaseBrowserStrategy class to reduce code duplication and improve maintainability. Key changes:
- Adds shared browser argument building and session management to base class
- Standardizes storage state handling across strategies
- Improves process cleanup and error handling
- Consolidates CDP URL management and container lifecycle

BREAKING CHANGE: Changes browser_mode="custom" to "cdp" for consistency
This commit is contained in:
UncleCode
2025-03-28 22:47:28 +08:00
parent 64f20ab44a
commit 3ff7eec8f3
12 changed files with 1102 additions and 671 deletions

View File

@@ -614,9 +614,9 @@ async def run_tests():
# return
# Run browser tests
# results.append(await test_docker_connect_mode())
# results.append(await test_docker_launch_mode())
# results.append(await test_docker_persistent_storage())
results.append(await test_docker_connect_mode())
results.append(await test_docker_launch_mode())
results.append(await test_docker_persistent_storage())
results.append(await test_docker_parallel_pages())
results.append(await test_docker_registry_reuse())

View File

@@ -77,7 +77,7 @@ async def test_builtin_browser_creation():
# Step 4: Get browser info from the strategy
print(f"\n{INFO}4. Getting browser information{RESET}")
browser_info = manager._strategy.get_builtin_browser_info()
browser_info = manager._strategy.get_browser_info()
if browser_info:
print(f"{SUCCESS}Browser info retrieved:{RESET}")
for key, value in browser_info.items():
@@ -205,7 +205,7 @@ async def test_multiple_managers():
# Step 1: Create first manager
print(f"\n{INFO}1. Creating first browser manager{RESET}")
browser_config1 = (BrowserConfig(browser_mode="builtin", headless=True),)
browser_config1 = BrowserConfig(browser_mode="builtin", headless=True)
manager1 = BrowserManager(browser_config=browser_config1, logger=logger)
# Step 2: Create second manager
@@ -781,15 +781,16 @@ async def main():
# await manager.close()
# Run multiple managers test
# await test_multiple_managers()
await test_multiple_managers()
# Run performance scaling test
await test_performance_scaling()
# Run cleanup test
# await cleanup_browsers()
await cleanup_browsers()
# Run edge cases test
# await test_edge_cases()
await test_edge_cases()
print(f"\n{SUCCESS}All tests completed!{RESET}")

View File

@@ -25,6 +25,7 @@ async def test_cdp_launch_connect():
browser_config = BrowserConfig(
use_managed_browser=True,
browser_mode="cdp",
headless=True
)
@@ -70,8 +71,8 @@ async def test_cdp_with_user_data_dir():
logger.info(f"Created temporary user data directory: {user_data_dir}", tag="TEST")
browser_config = BrowserConfig(
use_managed_browser=True,
headless=True,
browser_mode="cdp",
user_data_dir=user_data_dir
)
@@ -210,7 +211,7 @@ async def run_tests():
results = []
# results.append(await test_cdp_launch_connect())
# results.append(await test_cdp_with_user_data_dir())
results.append(await test_cdp_with_user_data_dir())
results.append(await test_cdp_session_management())
# Print summary

View File

@@ -6,6 +6,7 @@ and serve as functional tests.
import asyncio
import os
import re
import sys
# Add the project root to Python path if running directly
@@ -19,6 +20,53 @@ from crawl4ai.async_logger import AsyncLogger
# Create a logger for clear terminal output
logger = AsyncLogger(verbose=True, log_file=None)
async def test_start_close():
# Create browser config for standard Playwright
browser_config = BrowserConfig(
headless=True,
viewport_width=1280,
viewport_height=800
)
# Create browser manager with the config
manager = BrowserManager(browser_config=browser_config, logger=logger)
try:
for _ in range(4):
# Start the browser
await manager.start()
logger.info("Browser started successfully", tag="TEST")
# Get a page
page, context = await manager.get_page(CrawlerRunConfig())
logger.info("Got page successfully", tag="TEST")
# Navigate to a website
await page.goto("https://example.com")
logger.info("Navigated to example.com", tag="TEST")
# Get page title
title = await page.title()
logger.info(f"Page title: {title}", tag="TEST")
# Clean up
await manager.close()
logger.info("Browser closed successfully", tag="TEST")
await asyncio.sleep(1) # Wait for a moment before restarting
except Exception as e:
logger.error(f"Test failed: {str(e)}", tag="TEST")
# Ensure cleanup
try:
await manager.close()
except:
pass
return False
return True
async def test_playwright_basic():
"""Test basic Playwright browser functionality."""
logger.info("Testing standard Playwright browser", tag="TEST")
@@ -248,9 +296,10 @@ async def run_tests():
"""Run all tests sequentially."""
results = []
results.append(await test_playwright_basic())
results.append(await test_playwright_text_mode())
results.append(await test_playwright_context_reuse())
# results.append(await test_start_close())
# results.append(await test_playwright_basic())
# results.append(await test_playwright_text_mode())
# results.append(await test_playwright_context_reuse())
results.append(await test_playwright_session_management())
# Print summary