""" Test script for BrowserSession.start() method to ensure proper initialization, concurrency handling, and error handling. Tests cover: - Calling .start() on a session that's already started - Simultaneously calling .start() from two parallel coroutines - Calling .start() on a session that's started but has a closed browser connection - Calling .close() on a session that hasn't been started yet """ import asyncio import logging import pytest from browser_use.browser.profile import ( BROWSERUSE_DEFAULT_CHANNEL, BrowserChannel, BrowserProfile, ) from browser_use.browser.session import BrowserSession from browser_use.config import CONFIG # Set up test logging logger = logging.getLogger('browser_session_start_tests') # logger.setLevel(logging.DEBUG) # run with pytest -k test_user_data_dir_not_allowed_to_corrupt_default_profile class TestBrowserSessionStart: """Tests for BrowserSession.start() method initialization and concurrency.""" @pytest.fixture(scope='module') async def browser_profile(self): """Create and provide a BrowserProfile with headless mode.""" profile = BrowserProfile(headless=True, user_data_dir=None, keep_alive=False) yield profile @pytest.fixture(scope='function') async def browser_session(self, browser_profile): """Create a BrowserSession instance without starting it.""" session = BrowserSession(browser_profile=browser_profile) yield session await session.kill() async def test_start_already_started_session(self, browser_session): """Test calling .start() on a session that's already started.""" # logger.info('Testing start on already started session') # Start the session for the first time await browser_session.start() assert browser_session._cdp_client_root is not None # Start the session again - should return immediately without re-initialization await browser_session.start() assert browser_session._cdp_client_root is not None # @pytest.mark.skip(reason="Race condition - DOMWatchdog tries to inject scripts into tab that's being closed") # async def test_page_lifecycle_management(self, browser_session: BrowserSession): # """Test session handles page lifecycle correctly.""" # # logger.info('Testing page lifecycle management') # # Start the session and get initial state # await browser_session.start() # initial_tabs = await browser_session.get_tabs() # initial_count = len(initial_tabs) # # Get current tab info # current_url = await browser_session.get_current_page_url() # assert current_url is not None # # Get current tab ID # current_tab_id = browser_session.agent_focus.target_id if browser_session.agent_focus else None # assert current_tab_id is not None # # Close the current tab using the event system # from browser_use.browser.events import CloseTabEvent # close_event = browser_session.event_bus.dispatch(CloseTabEvent(target_id=current_tab_id)) # await close_event # # Operations should still work - may create new page or use existing # tabs_after_close = await browser_session.get_tabs() # assert isinstance(tabs_after_close, list) # # Create a new tab explicitly # event = browser_session.event_bus.dispatch(NavigateToUrlEvent(url='about:blank', new_tab=True)) # await event # await event.event_result(raise_if_any=True, raise_if_none=False) # # Should have at least one tab now # final_tabs = await browser_session.get_tabs() # assert len(final_tabs) >= 1 async def test_user_data_dir_not_allowed_to_corrupt_default_profile(self): """Test user_data_dir handling for different browser channels and version mismatches.""" # Test 1: Chromium with default user_data_dir and default channel should work fine session = BrowserSession( browser_profile=BrowserProfile( headless=True, user_data_dir=CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR, channel=BROWSERUSE_DEFAULT_CHANNEL, # chromium keep_alive=False, ), ) try: await session.start() assert session._cdp_client_root is not None # Verify the user_data_dir wasn't changed assert session.browser_profile.user_data_dir == CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR finally: await session.kill() # Test 2: Chrome with default user_data_dir should change dir AND copy to temp profile2 = BrowserProfile( headless=True, user_data_dir=CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR, channel=BrowserChannel.CHROME, keep_alive=False, ) # The validator should have changed the user_data_dir to avoid corruption # And then _copy_profile copies it to a temp directory (Chrome only) assert profile2.user_data_dir != CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR assert 'browser-use-user-data-dir-' in str(profile2.user_data_dir) # Test 3: Edge with default user_data_dir should also change profile3 = BrowserProfile( headless=True, user_data_dir=CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR, channel=BrowserChannel.MSEDGE, keep_alive=False, ) assert profile3.user_data_dir != CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR assert profile3.user_data_dir == CONFIG.BROWSER_USE_DEFAULT_USER_DATA_DIR.parent / 'default-msedge' assert 'browser-use-user-data-dir-' not in str(profile3.user_data_dir) class TestBrowserSessionReusePatterns: """Tests for all browser re-use patterns documented in docs/customize/real-browser.mdx""" async def test_sequential_agents_same_profile_different_browser(self, mock_llm): """Test Sequential Agents, Same Profile, Different Browser pattern""" from browser_use import Agent from browser_use.browser.profile import BrowserProfile # Create a reusable profile reused_profile = BrowserProfile( user_data_dir=None, # Use temp dir for testing headless=True, ) # First agent agent1 = Agent( task='The first task...', llm=mock_llm, browser_profile=reused_profile, ) await agent1.run() # Verify first agent's session is closed assert agent1.browser_session is not None assert not agent1.browser_session._cdp_client_root is not None # Second agent with same profile agent2 = Agent( task='The second task...', llm=mock_llm, browser_profile=reused_profile, # Disable memory for tests ) await agent2.run() # Verify second agent created a new session assert agent2.browser_session is not None assert agent1.browser_session is not agent2.browser_session assert not agent2.browser_session._cdp_client_root is not None async def test_sequential_agents_same_profile_same_browser(self, mock_llm): """Test Sequential Agents, Same Profile, Same Browser pattern""" from browser_use import Agent, BrowserSession # Create a reusable session with keep_alive reused_session = BrowserSession( browser_profile=BrowserProfile( user_data_dir=None, # Use temp dir for testing headless=True, keep_alive=True, # Don't close browser after agent.run() ), ) try: # Start the session manually (agents will reuse this initialized session) await reused_session.start() # First agent agent1 = Agent( task='The first task...', llm=mock_llm, browser_session=reused_session, # Disable memory for tests ) await agent1.run() # Verify session is still alive assert reused_session._cdp_client_root is not None # Second agent reusing the same session agent2 = Agent( task='The second task...', llm=mock_llm, browser_session=reused_session, # Disable memory for tests ) await agent2.run() # Verify same browser was used (using __eq__ to check browser_pid, cdp_url) assert agent1.browser_session == agent2.browser_session assert agent1.browser_session == reused_session assert reused_session._cdp_client_root is not None finally: await reused_session.kill() class TestBrowserSessionEventSystem: """Tests for the new event system integration in BrowserSession.""" @pytest.fixture(scope='function') async def browser_session(self): """Create a BrowserSession instance for event system testing.""" profile = BrowserProfile(headless=True, user_data_dir=None, keep_alive=False) session = BrowserSession(browser_profile=profile) yield session await session.kill() async def test_event_bus_initialization(self, browser_session): """Test that event bus is properly initialized with unique name.""" # Event bus should be created during __init__ assert browser_session.event_bus is not None assert browser_session.event_bus.name.startswith('EventBus_') # Event bus name format may vary, just check it exists async def test_event_handlers_registration(self, browser_session: BrowserSession): """Test that event handlers are properly registered.""" # Attach all watchdogs to register their handlers await browser_session.attach_all_watchdogs() # Check that handlers are registered in the event bus from browser_use.browser.events import ( BrowserStartEvent, BrowserStateRequestEvent, BrowserStopEvent, ClickElementEvent, CloseTabEvent, ScreenshotEvent, ScrollEvent, TypeTextEvent, ) # These event types should have handlers registered event_types_with_handlers = [ BrowserStartEvent, BrowserStopEvent, ClickElementEvent, TypeTextEvent, ScrollEvent, CloseTabEvent, BrowserStateRequestEvent, ScreenshotEvent, ] for event_type in event_types_with_handlers: handlers = browser_session.event_bus.handlers.get(event_type.__name__, []) assert len(handlers) > 0, f'No handlers registered for {event_type.__name__}' async def test_direct_event_dispatching(self, browser_session): """Test direct event dispatching without using the public API.""" from browser_use.browser.events import BrowserConnectedEvent, BrowserStartEvent # Dispatch BrowserStartEvent directly start_event = browser_session.event_bus.dispatch(BrowserStartEvent()) # Wait for event to complete await start_event # Check if BrowserConnectedEvent was dispatched assert browser_session._cdp_client_root is not None # Check event history event_history = list(browser_session.event_bus.event_history.values()) assert len(event_history) >= 2 # BrowserStartEvent + BrowserConnectedEvent + others # Find the BrowserConnectedEvent in history started_events = [e for e in event_history if isinstance(e, BrowserConnectedEvent)] assert len(started_events) >= 1 assert started_events[0].cdp_url is not None async def test_event_system_error_handling(self, browser_session): """Test error handling in event system.""" from browser_use.browser.events import BrowserStartEvent # Create session with invalid CDP URL to trigger error error_session = BrowserSession( browser_profile=BrowserProfile(headless=True), cdp_url='http://localhost:99999', # Invalid port ) try: # Dispatch start event directly - should trigger error handling start_event = error_session.event_bus.dispatch(BrowserStartEvent()) # The event bus catches and logs the error, but the event awaits successfully await start_event # The session should not be initialized due to the error assert error_session._cdp_client_root is None, 'Session should not be initialized after connection error' # Verify the error was logged in the event history (good enough for error handling test) assert len(error_session.event_bus.event_history) > 0, 'Event should be tracked even with errors' finally: await error_session.kill() async def test_concurrent_event_dispatching(self, browser_session: BrowserSession): """Test that concurrent events are handled properly.""" from browser_use.browser.events import ScreenshotEvent # Start browser first await browser_session.start() # Dispatch multiple events concurrently screenshot_event1 = browser_session.event_bus.dispatch(ScreenshotEvent()) screenshot_event2 = browser_session.event_bus.dispatch(ScreenshotEvent()) # Both should complete successfully results = await asyncio.gather(screenshot_event1, screenshot_event2, return_exceptions=True) # Check that no exceptions were raised for result in results: assert not isinstance(result, Exception), f'Event failed with: {result}' # async def test_many_parallel_browser_sessions(self): # """Test spawning 12 parallel browser_sessions with different settings and ensure they all work""" # from browser_use import BrowserSession # browser_sessions = [] # for i in range(3): # browser_sessions.append( # BrowserSession( # browser_profile=BrowserProfile( # user_data_dir=None, # headless=True, # keep_alive=True, # ), # ) # ) # for i in range(3): # browser_sessions.append( # BrowserSession( # browser_profile=BrowserProfile( # user_data_dir=Path(tempfile.mkdtemp(prefix=f'browseruse-tmp-{i}')), # headless=True, # keep_alive=True, # ), # ) # ) # for i in range(3): # browser_sessions.append( # BrowserSession( # browser_profile=BrowserProfile( # user_data_dir=None, # headless=True, # keep_alive=False, # ), # ) # ) # for i in range(3): # browser_sessions.append( # BrowserSession( # browser_profile=BrowserProfile( # user_data_dir=Path(tempfile.mkdtemp(prefix=f'browseruse-tmp-{i}')), # headless=True, # keep_alive=False, # ), # ) # ) # print('Starting many parallel browser sessions...') # await asyncio.gather(*[browser_session.start() for browser_session in browser_sessions]) # print('Ensuring all parallel browser sessions are connected and usable...') # new_tab_tasks = [] # for browser_session in browser_sessions: # assert browser_session._cdp_client_root is not None # assert browser_session._cdp_client_root is not None # new_tab_tasks.append(browser_session.create_new_tab('chrome://version')) # await asyncio.gather(*new_tab_tasks) # print('killing every 3rd browser_session to test parallel shutdown') # kill_tasks = [] # for i in range(0, len(browser_sessions), 3): # kill_tasks.append(browser_sessions[i].kill()) # browser_sessions[i] = None # results = await asyncio.gather(*kill_tasks, return_exceptions=True) # # Check that no exceptions were raised during cleanup # for i, result in enumerate(results): # if isinstance(result, Exception): # print(f'Warning: Browser session kill raised exception: {type(result).__name__}: {result}') # print('ensuring the remaining browser_sessions are still connected and usable') # new_tab_tasks = [] # screenshot_tasks = [] # for browser_session in filter(bool, browser_sessions): # assert browser_session._cdp_client_root is not None # assert browser_session._cdp_client_root is not None # new_tab_tasks.append(browser_session.create_new_tab('chrome://version')) # screenshot_tasks.append(browser_session.take_screenshot()) # await asyncio.gather(*new_tab_tasks) # await asyncio.gather(*screenshot_tasks) # kill_tasks = [] # print('killing the remaining browser_sessions') # for browser_session in filter(bool, browser_sessions): # kill_tasks.append(browser_session.kill()) # results = await asyncio.gather(*kill_tasks, return_exceptions=True) # # Check that no exceptions were raised during cleanup # for i, result in enumerate(results): # if isinstance(result, Exception): # print(f'Warning: Browser session kill raised exception: {type(result).__name__}: {result}')