mirror of
https://github.com/browser-use/browser-use
synced 2026-05-13 17:56:35 +02:00
520 lines
19 KiB
Python
520 lines
19 KiB
Python
"""Test navigation events are emitted properly in all cases."""
|
|
|
|
import asyncio
|
|
import time
|
|
from typing import cast
|
|
|
|
import pytest
|
|
|
|
from browser_use.browser.events import (
|
|
ClickElementEvent,
|
|
NavigateToUrlEvent,
|
|
NavigationCompleteEvent,
|
|
NavigationStartedEvent,
|
|
TabCreatedEvent,
|
|
)
|
|
from browser_use.browser.profile import BrowserProfile
|
|
from browser_use.browser.session import BrowserSession
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_navigation_events_fast_page_load(httpserver):
|
|
"""Test navigation events for page that loads easily/normally within 1s."""
|
|
# Set up a fast endpoint
|
|
httpserver.expect_request('/fast').respond_with_data(
|
|
'<html><head><title>Fast Page</title></head><body><h1>Fast Loading Page</h1></body></html>',
|
|
status=200,
|
|
content_type='text/html',
|
|
)
|
|
fast_url = httpserver.url_for('/fast')
|
|
|
|
profile = BrowserProfile(headless=True)
|
|
session = BrowserSession(browser_profile=profile)
|
|
|
|
# Track navigation events
|
|
navigation_started_events = []
|
|
navigation_complete_events = []
|
|
session.event_bus.on(NavigationStartedEvent, lambda e: navigation_started_events.append(e))
|
|
session.event_bus.on(NavigationCompleteEvent, lambda e: navigation_complete_events.append(e))
|
|
|
|
try:
|
|
# Start browser
|
|
await session.start()
|
|
|
|
# Navigate to fast page
|
|
start_time = time.time()
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=fast_url))
|
|
|
|
# Wait for navigation to complete
|
|
nav_complete: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
)
|
|
end_time = time.time()
|
|
|
|
# Verify navigation completed quickly (within 1s)
|
|
assert (end_time - start_time) < 1.0, 'Navigation should complete quickly'
|
|
|
|
# Verify NavigationStartedEvent was emitted
|
|
assert len(navigation_started_events) >= 1, 'Should have NavigationStartedEvent'
|
|
nav_started = navigation_started_events[-1]
|
|
assert nav_started.url == fast_url
|
|
assert nav_started.tab_index >= 0
|
|
|
|
# Verify NavigationCompleteEvent was emitted with success
|
|
assert len(navigation_complete_events) >= 1, 'Should have NavigationCompleteEvent'
|
|
assert nav_complete.url == fast_url
|
|
assert nav_complete.tab_index >= 0
|
|
assert nav_complete.status == 200, 'Should have successful HTTP status'
|
|
assert nav_complete.error_message is None, 'Should have no error message'
|
|
assert nav_complete.loading_status is None, 'Should have no loading status issues'
|
|
|
|
finally:
|
|
await session.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_navigation_events_slow_page_with_timeout(httpserver):
|
|
"""Test navigation events for page that takes >10s to load and times out."""
|
|
|
|
# Set up a slow endpoint that takes longer than we want to wait
|
|
def slow_handler(request):
|
|
time.sleep(5.0) # 5 seconds - longer than our timeout
|
|
from werkzeug import Response
|
|
|
|
return Response('<html><body>Finally loaded</body></html>', status=200)
|
|
|
|
httpserver.expect_request('/slow').respond_with_handler(slow_handler)
|
|
slow_url = httpserver.url_for('/slow')
|
|
|
|
# Create profile with shorter timeout for faster testing
|
|
profile = BrowserProfile(
|
|
headless=True,
|
|
maximum_wait_page_load_time=2.0, # 2 second timeout for network monitoring
|
|
wait_for_network_idle_page_load_time=0.5, # 0.5 second network idle
|
|
)
|
|
session = BrowserSession(browser_profile=profile)
|
|
|
|
# Track navigation events
|
|
navigation_started_events = []
|
|
navigation_complete_events = []
|
|
session.event_bus.on(NavigationStartedEvent, lambda e: navigation_started_events.append(e))
|
|
session.event_bus.on(NavigationCompleteEvent, lambda e: navigation_complete_events.append(e))
|
|
|
|
try:
|
|
# Start browser
|
|
await session.start()
|
|
|
|
# Navigate to slow page
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=slow_url))
|
|
|
|
# Wait for navigation to timeout and complete with error
|
|
nav_complete: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=10.0)
|
|
)
|
|
|
|
# Verify NavigationStartedEvent was emitted
|
|
assert len(navigation_started_events) >= 1, 'Should have NavigationStartedEvent'
|
|
nav_started = navigation_started_events[-1]
|
|
assert nav_started.url == slow_url
|
|
|
|
# Verify NavigationCompleteEvent was emitted
|
|
assert len(navigation_complete_events) >= 1, 'Should have NavigationCompleteEvent'
|
|
assert nav_complete.url == slow_url
|
|
|
|
# Either the navigation succeeded after timeout (page loaded) or it timed out
|
|
# Both are valid outcomes - what matters is that NavigationCompleteEvent was emitted
|
|
if nav_complete.error_message or nav_complete.loading_status:
|
|
# Navigation had issues (timeout/loading problems)
|
|
has_timeout_indicator = (
|
|
nav_complete.error_message
|
|
and ('timeout' in nav_complete.error_message.lower() or 'pending' in nav_complete.error_message.lower())
|
|
or nav_complete.loading_status
|
|
and (
|
|
'aborted' in nav_complete.loading_status.lower()
|
|
or 'pending' in nav_complete.loading_status.lower()
|
|
or 'requests' in nav_complete.loading_status.lower()
|
|
)
|
|
)
|
|
print(
|
|
f"Navigation had issues - error_message: '{nav_complete.error_message}', loading_status: '{nav_complete.loading_status}'"
|
|
)
|
|
assert has_timeout_indicator, (
|
|
f'Should indicate timeout/loading issues when navigation has problems. '
|
|
f"error_message='{nav_complete.error_message}', "
|
|
f"loading_status='{nav_complete.loading_status}'"
|
|
)
|
|
else:
|
|
# Navigation succeeded (slow but successful)
|
|
print('Navigation succeeded despite being slow')
|
|
assert nav_complete.status == 200, 'Successful navigation should have HTTP 200'
|
|
|
|
finally:
|
|
await session.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_navigation_events_history_pushstate(httpserver):
|
|
"""Test that history.pushState/popState behavior and that clicking links produces NavigationCompleteEvent."""
|
|
# Set up a page with JavaScript that manipulates history and contains navigation links
|
|
html_content = """
|
|
<html>
|
|
<head><title>History Test</title></head>
|
|
<body>
|
|
<h1>History and Navigation Test</h1>
|
|
<button id="push-btn" onclick="pushState()">Push State</button>
|
|
<button id="pop-btn" onclick="history.back()">Pop State</button>
|
|
<a id="same-tab-link" href="/target-page">Same Tab Link</a>
|
|
<a id="new-tab-link" href="/target-page" target="_blank">New Tab Link</a>
|
|
<div id="status">Initial state</div>
|
|
<script>
|
|
function pushState() {
|
|
history.pushState({page: 'new'}, 'New Page', '/new-path');
|
|
document.getElementById('status').textContent = 'Pushed new state';
|
|
}
|
|
|
|
window.addEventListener('popstate', function(event) {
|
|
document.getElementById('status').textContent = 'Popped state: ' + JSON.stringify(event.state);
|
|
});
|
|
</script>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
target_page = """
|
|
<html>
|
|
<head><title>Target Page</title></head>
|
|
<body><h1>Target Page Loaded</h1></body>
|
|
</html>
|
|
"""
|
|
|
|
httpserver.expect_request('/history-test').respond_with_data(html_content, status=200, content_type='text/html')
|
|
httpserver.expect_request('/target-page').respond_with_data(target_page, status=200, content_type='text/html')
|
|
|
|
test_url = httpserver.url_for('/history-test')
|
|
target_url = httpserver.url_for('/target-page')
|
|
|
|
profile = BrowserProfile(headless=True)
|
|
session = BrowserSession(browser_profile=profile)
|
|
|
|
# Track navigation events
|
|
navigation_complete_events = []
|
|
tab_created_events = []
|
|
session.event_bus.on(NavigationCompleteEvent, lambda e: navigation_complete_events.append(e))
|
|
session.event_bus.on(TabCreatedEvent, lambda e: tab_created_events.append(e))
|
|
|
|
try:
|
|
# Start browser
|
|
await session.start()
|
|
|
|
# Navigate to the test page
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=test_url))
|
|
await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
|
|
# Clear previous events to focus on history and link events
|
|
navigation_complete_events.clear()
|
|
tab_created_events.clear()
|
|
|
|
# Get browser state to interact with the page
|
|
state = await session.get_browser_state_with_recovery()
|
|
|
|
# Test 1: history.pushState (should NOT trigger NavigationCompleteEvent)
|
|
push_button_found = False
|
|
for idx, element in state.selector_map.items():
|
|
if hasattr(element, 'attributes') and element.attributes.get('id') == 'push-btn':
|
|
push_button_found = True
|
|
session.event_bus.dispatch(ClickElementEvent(index=idx))
|
|
break
|
|
|
|
assert push_button_found, 'Should find push state button'
|
|
|
|
# Wait for pushState to execute
|
|
await asyncio.sleep(1.0)
|
|
|
|
# Verify history.pushState does NOT trigger NavigationCompleteEvent (expected behavior)
|
|
assert len(navigation_complete_events) == 0, 'history.pushState should not trigger NavigationCompleteEvent'
|
|
|
|
# Test 2: Same tab link click (should trigger NavigationCompleteEvent)
|
|
state = await session.get_browser_state_with_recovery()
|
|
same_tab_link_found = False
|
|
|
|
for idx, element in state.selector_map.items():
|
|
if hasattr(element, 'attributes') and element.attributes.get('id') == 'same-tab-link':
|
|
same_tab_link_found = True
|
|
session.event_bus.dispatch(ClickElementEvent(index=idx))
|
|
break
|
|
|
|
assert same_tab_link_found, 'Should find same tab link'
|
|
|
|
# Wait for navigation to complete
|
|
nav_complete: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=10.0)
|
|
)
|
|
|
|
assert nav_complete.url == target_url, f'Should navigate to {target_url}'
|
|
assert nav_complete.error_message is None, 'Link navigation should succeed'
|
|
|
|
# Test 3: New tab link click (may trigger TabCreatedEvent and/or NavigationCompleteEvent)
|
|
# Navigate back to main page first
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=test_url))
|
|
await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
|
|
# Clear events
|
|
navigation_complete_events.clear()
|
|
tab_created_events.clear()
|
|
|
|
state = await session.get_browser_state_with_recovery()
|
|
new_tab_link_found = False
|
|
|
|
for idx, element in state.selector_map.items():
|
|
if hasattr(element, 'attributes') and element.attributes.get('id') == 'new-tab-link':
|
|
new_tab_link_found = True
|
|
# Use new_tab=True parameter to properly trigger new tab behavior
|
|
session.event_bus.dispatch(ClickElementEvent(index=idx, new_tab=True))
|
|
break
|
|
|
|
if new_tab_link_found:
|
|
# Wait for either new tab creation or navigation
|
|
await asyncio.sleep(2.0) # Give time for tab creation and navigation
|
|
|
|
# Should have either tab creation or navigation event (or both)
|
|
has_new_tab_activity = len(tab_created_events) > 0 or len(navigation_complete_events) > 0
|
|
assert has_new_tab_activity, 'New tab link should trigger tab creation or navigation'
|
|
|
|
if navigation_complete_events:
|
|
nav_complete = navigation_complete_events[-1]
|
|
assert target_url in nav_complete.url or nav_complete.error_message is None, (
|
|
f'New tab navigation should succeed, got: {nav_complete.error_message}'
|
|
)
|
|
|
|
# Test 4: Verify that normal navigation still works after history manipulation
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url='data:text/html,<h1>After history test</h1>'))
|
|
nav_complete_final: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
)
|
|
|
|
assert nav_complete_final.url == 'data:text/html,<h1>After history test</h1>'
|
|
assert nav_complete_final.error_message is None
|
|
|
|
finally:
|
|
await session.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_navigation_events_link_clicks(httpserver):
|
|
"""Test that clicking links (same tab and new tab) triggers NavigationCompleteEvent."""
|
|
# Set up pages with different types of links
|
|
main_page = """
|
|
<html>
|
|
<head><title>Link Test</title></head>
|
|
<body>
|
|
<h1>Link Click Test</h1>
|
|
<a id="same-tab-link" href="/target-page">Same Tab Link</a>
|
|
<a id="new-tab-link" href="/target-page" target="_blank">New Tab Link</a>
|
|
<a id="js-navigation" href="#" onclick="window.location.href='/js-target'; return false;">JS Navigation</a>
|
|
</body>
|
|
</html>
|
|
"""
|
|
|
|
target_page = """
|
|
<html>
|
|
<head><title>Target Page</title></head>
|
|
<body><h1>Target Page Loaded</h1></body>
|
|
</html>
|
|
"""
|
|
|
|
js_target_page = """
|
|
<html>
|
|
<head><title>JS Target</title></head>
|
|
<body><h1>JavaScript Navigation Target</h1></body>
|
|
</html>
|
|
"""
|
|
|
|
httpserver.expect_request('/link-test').respond_with_data(main_page, status=200, content_type='text/html')
|
|
httpserver.expect_request('/target-page').respond_with_data(target_page, status=200, content_type='text/html')
|
|
httpserver.expect_request('/js-target').respond_with_data(js_target_page, status=200, content_type='text/html')
|
|
|
|
main_url = httpserver.url_for('/link-test')
|
|
target_url = httpserver.url_for('/target-page')
|
|
js_target_url = httpserver.url_for('/js-target')
|
|
|
|
profile = BrowserProfile(headless=True)
|
|
session = BrowserSession(browser_profile=profile)
|
|
|
|
# Track navigation and tab events
|
|
navigation_complete_events = []
|
|
tab_created_events = []
|
|
session.event_bus.on(NavigationCompleteEvent, lambda e: navigation_complete_events.append(e))
|
|
session.event_bus.on(TabCreatedEvent, lambda e: tab_created_events.append(e))
|
|
|
|
try:
|
|
# Start browser
|
|
await session.start()
|
|
|
|
# Navigate to the main page
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=main_url))
|
|
await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
|
|
# Clear events to focus on link clicks
|
|
navigation_complete_events.clear()
|
|
|
|
# Test 1: Same tab link click
|
|
state = await session.get_browser_state_with_recovery()
|
|
same_tab_link_found = False
|
|
|
|
for idx, element in state.selector_map.items():
|
|
if hasattr(element, 'attributes') and element.attributes.get('id') == 'same-tab-link':
|
|
same_tab_link_found = True
|
|
session.event_bus.dispatch(ClickElementEvent(index=idx))
|
|
break
|
|
|
|
assert same_tab_link_found, 'Should find same tab link'
|
|
|
|
# Wait for navigation to complete
|
|
nav_complete: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
)
|
|
|
|
assert nav_complete.url == target_url, f'Should navigate to {target_url}'
|
|
assert nav_complete.error_message is None, 'Link navigation should succeed'
|
|
|
|
# Test 2: New tab link click (if supported)
|
|
# Navigate back to main page first
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=main_url))
|
|
await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
|
|
# Clear events
|
|
navigation_complete_events.clear()
|
|
tab_created_events.clear()
|
|
|
|
state = await session.get_browser_state_with_recovery()
|
|
new_tab_link_found = False
|
|
|
|
for idx, element in state.selector_map.items():
|
|
if hasattr(element, 'attributes') and element.attributes.get('id') == 'new-tab-link':
|
|
new_tab_link_found = True
|
|
session.event_bus.dispatch(ClickElementEvent(index=idx, new_tab=True))
|
|
break
|
|
|
|
if new_tab_link_found:
|
|
# Wait for either new tab creation or navigation
|
|
await asyncio.sleep(2.0) # Give time for tab creation and navigation
|
|
|
|
# Should have either tab creation or navigation event (or both)
|
|
has_new_tab_activity = len(tab_created_events) > 0 or len(navigation_complete_events) > 0
|
|
assert has_new_tab_activity, 'New tab link should trigger tab creation or navigation'
|
|
|
|
if navigation_complete_events:
|
|
nav_complete = navigation_complete_events[-1]
|
|
assert target_url in nav_complete.url or nav_complete.error_message is None
|
|
|
|
# Test 3: JavaScript navigation
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url=main_url))
|
|
await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
|
|
navigation_complete_events.clear()
|
|
|
|
state = await session.get_browser_state_with_recovery()
|
|
js_link_found = False
|
|
|
|
for idx, element in state.selector_map.items():
|
|
if hasattr(element, 'attributes') and element.attributes.get('id') == 'js-navigation':
|
|
js_link_found = True
|
|
session.event_bus.dispatch(ClickElementEvent(index=idx))
|
|
break
|
|
|
|
if js_link_found:
|
|
# Wait for JavaScript navigation to complete
|
|
nav_complete = cast(NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0))
|
|
|
|
assert nav_complete.url == js_target_url, 'JS navigation should work'
|
|
assert nav_complete.error_message is None, 'JS navigation should succeed'
|
|
|
|
finally:
|
|
await session.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_navigation_timeout_event_dispatch():
|
|
"""Test that NavigateToUrlEvent with timeout_ms properly dispatches NavigationCompleteEvent on timeout."""
|
|
profile = BrowserProfile(headless=True)
|
|
session = BrowserSession(browser_profile=profile)
|
|
|
|
# Track navigation events
|
|
navigation_complete_events = []
|
|
session.event_bus.on(NavigationCompleteEvent, lambda e: navigation_complete_events.append(e))
|
|
|
|
try:
|
|
# Start browser
|
|
await session.start()
|
|
|
|
# Navigate with a very short timeout to a valid but slow URL
|
|
session.event_bus.dispatch(
|
|
NavigateToUrlEvent(
|
|
url='data:text/html,<h1>Should timeout</h1>', # Use a data URL that should work
|
|
timeout_ms=1, # 1ms timeout - should definitely timeout
|
|
)
|
|
)
|
|
|
|
# Wait for navigation to timeout and complete with error
|
|
nav_complete: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
)
|
|
|
|
# Verify NavigationCompleteEvent indicates timeout
|
|
assert nav_complete.error_message is not None, 'Should have error message for timeout'
|
|
assert 'timed out' in nav_complete.error_message.lower(), f'Error should mention timed out: {nav_complete.error_message}'
|
|
assert nav_complete.loading_status is not None, 'Should have loading status'
|
|
assert 'timeout' in nav_complete.loading_status.lower(), (
|
|
f'Loading status should mention timeout: {nav_complete.loading_status}'
|
|
)
|
|
|
|
# Verify specific timeout details
|
|
assert '1ms' in nav_complete.error_message, 'Should mention the specific timeout duration'
|
|
|
|
finally:
|
|
await session.stop()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_navigation_error_recovery():
|
|
"""Test that navigation errors are properly reported and don't break subsequent navigation."""
|
|
profile = BrowserProfile(headless=True)
|
|
session = BrowserSession(browser_profile=profile)
|
|
|
|
# Track navigation events
|
|
navigation_complete_events = []
|
|
session.event_bus.on(NavigationCompleteEvent, lambda e: navigation_complete_events.append(e))
|
|
|
|
try:
|
|
# Start browser
|
|
await session.start()
|
|
|
|
# Try to navigate to invalid URL
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url='invalid://not-a-real-url'))
|
|
|
|
# Wait for navigation to fail
|
|
nav_complete: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
)
|
|
|
|
# Verify NavigationCompleteEvent indicates error
|
|
assert nav_complete.error_message is not None, 'Should have error message for invalid URL'
|
|
assert nav_complete.url == 'invalid://not-a-real-url'
|
|
|
|
# Clear events
|
|
navigation_complete_events.clear()
|
|
|
|
# Verify that subsequent navigation still works
|
|
session.event_bus.dispatch(NavigateToUrlEvent(url='data:text/html,<h1>Recovery Test</h1>'))
|
|
|
|
nav_complete_recovery: NavigationCompleteEvent = cast(
|
|
NavigationCompleteEvent, await session.event_bus.expect(NavigationCompleteEvent, timeout=5.0)
|
|
)
|
|
|
|
# Verify recovery navigation succeeded
|
|
assert nav_complete_recovery.url == 'data:text/html,<h1>Recovery Test</h1>'
|
|
assert nav_complete_recovery.error_message is None, 'Recovery navigation should succeed'
|
|
|
|
finally:
|
|
await session.stop()
|