mirror of
https://github.com/browser-use/browser-use
synced 2026-05-06 17:52:15 +02:00
407 lines
15 KiB
Python
407 lines
15 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
from pytest_httpserver import HTTPServer
|
|
|
|
from browser_use.agent.views import ActionModel, ActionResult
|
|
from browser_use.browser import BrowserSession
|
|
from browser_use.browser.profile import BrowserProfile
|
|
from browser_use.tools.service import Tools
|
|
from browser_use.tools.views import GoToUrlAction
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def http_server():
|
|
"""Create and provide a test HTTP server that serves static content."""
|
|
server = HTTPServer()
|
|
server.start()
|
|
|
|
# Add routes for common test pages
|
|
server.expect_request('/').respond_with_data(
|
|
'<html><head><title>Test Home Page</title></head><body><h1>Test Home Page</h1><p>Welcome to the test site</p></body></html>',
|
|
content_type='text/html',
|
|
)
|
|
|
|
server.expect_request('/page1').respond_with_data(
|
|
'<html><head><title>Test Page 1</title></head><body><h1>Test Page 1</h1><p>This is test page 1</p></body></html>',
|
|
content_type='text/html',
|
|
)
|
|
|
|
server.expect_request('/page2').respond_with_data(
|
|
'<html><head><title>Test Page 2</title></head><body><h1>Test Page 2</h1><p>This is test page 2</p></body></html>',
|
|
content_type='text/html',
|
|
)
|
|
|
|
yield server
|
|
server.stop()
|
|
|
|
|
|
@pytest.fixture(scope='session')
|
|
def base_url(http_server):
|
|
"""Return the base URL for the test HTTP server."""
|
|
return f'http://{http_server.host}:{http_server.port}'
|
|
|
|
|
|
@pytest.fixture(scope='module')
|
|
async def browser_session():
|
|
"""Create and provide a Browser instance with security disabled."""
|
|
profile = BrowserProfile(headless=True, disable_security=True, cross_origin_iframes=False)
|
|
session = BrowserSession(browser_profile=profile)
|
|
await session.start()
|
|
yield session
|
|
await session.kill()
|
|
|
|
|
|
@pytest.fixture
|
|
def tools():
|
|
"""Create and provide a Tools instance."""
|
|
return Tools()
|
|
|
|
|
|
class TestNavigateToUrlEvent:
|
|
"""Test NavigateToUrlEvent and go_to_url action functionality."""
|
|
|
|
async def test_go_to_url_action(self, tools, browser_session: BrowserSession, base_url):
|
|
"""Test that GoToUrlAction navigates to the specified URL and test both state summary methods."""
|
|
# Test successful navigation to a valid page
|
|
action_data = {'go_to_url': GoToUrlAction(url=f'{base_url}/page1', new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
action_model = GoToUrlActionModel(**action_data)
|
|
result = await tools.act(action_model, browser_session)
|
|
|
|
# Verify the successful navigation result
|
|
assert isinstance(result, ActionResult)
|
|
assert result.extracted_content is not None
|
|
assert f'Navigated to {base_url}' in result.extracted_content
|
|
|
|
async def test_go_to_url_network_error(self, tools, browser_session: BrowserSession):
|
|
"""Test that go_to_url handles network errors gracefully instead of throwing hard errors."""
|
|
# Create action model for go_to_url with an invalid domain
|
|
action_data = {'go_to_url': GoToUrlAction(url='https://www.nonexistentdndbeyond.com/', new_tab=False)}
|
|
|
|
# Create the ActionModel instance
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
action_model = GoToUrlActionModel(**action_data)
|
|
|
|
# Execute the action - should return soft error instead of throwing
|
|
result = await tools.act(action_model, browser_session)
|
|
|
|
# Verify the result
|
|
assert isinstance(result, ActionResult)
|
|
# The navigation should fail with an error for non-existent domain
|
|
|
|
# Test that get_state_summary works
|
|
try:
|
|
await browser_session.get_browser_state_summary()
|
|
assert False, 'Expected throw error when navigating to non-existent page'
|
|
except Exception as e:
|
|
pass
|
|
|
|
# Test that browser state recovery works after error
|
|
summary = await browser_session.get_browser_state_summary(include_screenshot=False)
|
|
assert summary is not None
|
|
|
|
async def test_navigate_to_url_event_directly(self, browser_session, base_url):
|
|
"""Test NavigateToUrlEvent directly through the event bus."""
|
|
from browser_use.browser.events import NavigateToUrlEvent
|
|
|
|
# Test navigation to a valid URL
|
|
event = browser_session.event_bus.dispatch(NavigateToUrlEvent(url=f'{base_url}/page1'))
|
|
result = await asyncio.wait_for(event, timeout=3.0)
|
|
# NavigateToUrlEvent handlers don't return values, just wait for completion
|
|
assert result is not None
|
|
|
|
# Wait a bit for navigation to complete
|
|
await asyncio.sleep(0.5)
|
|
|
|
# Verify we're on the correct page
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert f'{base_url}/page1' in current_url
|
|
|
|
async def test_go_to_url_new_tab(self, tools, browser_session, base_url):
|
|
"""Test that GoToUrlAction with new_tab=True opens URL in a new tab."""
|
|
# Get initial tab count
|
|
initial_tabs = await browser_session.get_tabs()
|
|
initial_tab_count = len(initial_tabs)
|
|
|
|
# Navigate to URL in new tab
|
|
action_data = {'go_to_url': GoToUrlAction(url=f'{base_url}/page2', new_tab=True)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
result = await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
await asyncio.sleep(0.5)
|
|
|
|
# Verify result
|
|
assert isinstance(result, ActionResult)
|
|
assert result.extracted_content is not None
|
|
assert 'Opened new tab with url' in result.extracted_content or 'Navigated to' in result.extracted_content
|
|
|
|
# Verify new tab was created
|
|
final_tabs = await browser_session.get_tabs()
|
|
final_tab_count = len(final_tabs)
|
|
assert final_tab_count == initial_tab_count + 1
|
|
|
|
# Verify we're on the new page
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert f'{base_url}/page2' in current_url
|
|
|
|
async def test_navigate_javascript_url(self, tools, browser_session, base_url):
|
|
"""Test that javascript: URLs are handled appropriately."""
|
|
# Navigate to a normal page first
|
|
action_data = {'go_to_url': GoToUrlAction(url=f'{base_url}/page1', new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
|
|
# Try to navigate to javascript: URL (should be handled gracefully)
|
|
js_action = {'go_to_url': GoToUrlAction(url='javascript:alert("test")', new_tab=False)}
|
|
result = await tools.act(GoToUrlActionModel(**js_action), browser_session)
|
|
|
|
# Should either succeed or fail gracefully
|
|
assert isinstance(result, ActionResult)
|
|
|
|
async def test_navigate_data_url(self, tools, browser_session):
|
|
"""Test navigating to a data: URL."""
|
|
# Create a simple data URL
|
|
data_url = 'data:text/html,<html><head><title>Data URL Test</title></head><body><h1>Data URL Content</h1></body></html>'
|
|
|
|
action_data = {'go_to_url': GoToUrlAction(url=data_url, new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
result = await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
|
|
# Verify navigation
|
|
assert isinstance(result, ActionResult)
|
|
assert result.extracted_content is not None
|
|
|
|
# Verify we can get the page title using CDP
|
|
title = await browser_session.get_current_page_title()
|
|
assert title == 'Data URL Test'
|
|
|
|
async def test_navigate_with_hash(self, tools, browser_session, base_url, http_server):
|
|
"""Test navigating to URLs with hash fragments."""
|
|
# Add a page with anchors
|
|
http_server.expect_request('/page-with-anchors').respond_with_data(
|
|
"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head><title>Page with Anchors</title></head>
|
|
<body>
|
|
<h1 id="top">Top of Page</h1>
|
|
<div style="height: 2000px;">Content</div>
|
|
<h2 id="section1">Section 1</h2>
|
|
<div style="height: 1000px;">More content</div>
|
|
<h2 id="section2">Section 2</h2>
|
|
</body>
|
|
</html>
|
|
""",
|
|
content_type='text/html',
|
|
)
|
|
|
|
# Navigate to page with hash
|
|
action_data = {'go_to_url': GoToUrlAction(url=f'{base_url}/page-with-anchors#section1', new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
result = await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
|
|
# Verify navigation
|
|
assert isinstance(result, ActionResult)
|
|
assert result.extracted_content is not None
|
|
|
|
# Verify URL includes hash
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert '#section1' in current_url
|
|
|
|
async def test_navigate_with_query_params(self, tools, browser_session, base_url, http_server):
|
|
"""Test navigating to URLs with query parameters."""
|
|
# Add a page that shows query params
|
|
http_server.expect_request('/search').respond_with_data(
|
|
"""
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head><title>Search Page</title></head>
|
|
<body>
|
|
<h1>Search Results</h1>
|
|
<div id="query"></div>
|
|
<script>
|
|
const params = new URLSearchParams(window.location.search);
|
|
document.getElementById('query').textContent = 'Query: ' + params.get('q');
|
|
</script>
|
|
</body>
|
|
</html>
|
|
""",
|
|
content_type='text/html',
|
|
)
|
|
|
|
# Navigate with query parameters
|
|
action_data = {'go_to_url': GoToUrlAction(url=f'{base_url}/search?q=test+query&page=1', new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
result = await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
|
|
# Verify navigation
|
|
assert isinstance(result, ActionResult)
|
|
assert result.extracted_content is not None
|
|
|
|
# Verify URL includes query params
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert 'q=test+query' in current_url or 'q=test%20query' in current_url
|
|
assert 'page=1' in current_url
|
|
|
|
@pytest.mark.skip(reason='Tab count assertion failures - tab management logic changed')
|
|
async def test_navigate_multiple_tabs(self, tools, browser_session, base_url):
|
|
"""Test navigating in multiple tabs sequentially."""
|
|
# Navigate to first page in current tab
|
|
action1 = {'go_to_url': GoToUrlAction(url=f'{base_url}/page1', new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
await tools.act(GoToUrlActionModel(**action1), browser_session)
|
|
|
|
# Open second page in new tab
|
|
action2 = {'go_to_url': GoToUrlAction(url=f'{base_url}/page2', new_tab=True)}
|
|
await tools.act(GoToUrlActionModel(**action2), browser_session)
|
|
|
|
# Open home page in yet another new tab
|
|
action3 = {'go_to_url': GoToUrlAction(url=base_url, new_tab=True)}
|
|
await tools.act(GoToUrlActionModel(**action3), browser_session)
|
|
|
|
# Should have 3 tabs now
|
|
tabs = await browser_session.get_tabs()
|
|
assert len(tabs) == 3
|
|
|
|
# Current tab should be the last one opened
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert base_url in current_url and '/page' not in current_url
|
|
|
|
async def test_navigate_timeout_handling(self, tools, browser_session):
|
|
"""Test that navigation timeouts are handled gracefully."""
|
|
# Try to navigate to a URL that will likely timeout
|
|
# Using a private IP that's unlikely to respond
|
|
timeout_url = 'http://192.0.2.1:8080/timeout'
|
|
|
|
action_data = {'go_to_url': GoToUrlAction(url=timeout_url, new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
# This should complete without hanging indefinitely
|
|
result = await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
|
|
# Should get a result (possibly with error)
|
|
assert isinstance(result, ActionResult)
|
|
|
|
async def test_navigate_redirect(self, tools, browser_session, base_url, http_server):
|
|
"""Test navigating to a URL that redirects."""
|
|
# Add a redirect endpoint
|
|
http_server.expect_request('/redirect').respond_with_data(
|
|
'',
|
|
status=302,
|
|
headers={'Location': f'{base_url}/page2'},
|
|
)
|
|
|
|
# Navigate to redirect URL
|
|
action_data = {'go_to_url': GoToUrlAction(url=f'{base_url}/redirect', new_tab=False)}
|
|
|
|
class GoToUrlActionModel(ActionModel):
|
|
go_to_url: GoToUrlAction | None = None
|
|
|
|
result = await tools.act(GoToUrlActionModel(**action_data), browser_session)
|
|
|
|
# Verify navigation succeeded
|
|
assert isinstance(result, ActionResult)
|
|
assert result.extracted_content is not None
|
|
|
|
# Should end up on page2 after redirect
|
|
await asyncio.sleep(0.5) # Give redirect time to complete
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert '/page2' in current_url
|
|
|
|
async def test_navigate_to_url_event_with_new_tab_and_tab_created_event(self, browser_session, base_url):
|
|
"""Test NavigateToUrlEvent with new_tab=True and verify TabCreatedEvent is emitted."""
|
|
from browser_use.browser.events import NavigateToUrlEvent, TabCreatedEvent
|
|
|
|
initial_tabs = await browser_session.get_tabs()
|
|
initial_tab_count = len(initial_tabs)
|
|
|
|
# Navigate to URL in new tab via direct event
|
|
nav_event = browser_session.event_bus.dispatch(NavigateToUrlEvent(url=f'{base_url}/page2', new_tab=True))
|
|
await nav_event
|
|
|
|
# Verify new tab was created
|
|
final_tabs = await browser_session.get_tabs()
|
|
assert len(final_tabs) == initial_tab_count + 1
|
|
|
|
# Check that current page is the new tab
|
|
current_url = await browser_session.get_current_page_url()
|
|
assert f'{base_url}/page2' in current_url
|
|
|
|
# Check event history for TabCreatedEvent
|
|
event_history = list(browser_session.event_bus.event_history.values())
|
|
created_events = [e for e in event_history if isinstance(e, TabCreatedEvent)]
|
|
assert len(created_events) >= 1
|
|
|
|
async def test_navigate_with_new_tab_focuses_properly(self, browser_session):
|
|
"""Test that NavigateToUrlEvent with new_tab=True properly switches focus."""
|
|
from browser_use.browser.events import NavigateToUrlEvent
|
|
|
|
# Get initial state
|
|
initial_tabs = await browser_session.get_tabs()
|
|
initial_tabs_count = len(initial_tabs)
|
|
initial_url = await browser_session.get_current_page_url()
|
|
|
|
# Navigate to a URL in a new tab
|
|
nav_event = browser_session.event_bus.dispatch(NavigateToUrlEvent(url='https://example.com', new_tab=True))
|
|
await nav_event
|
|
|
|
# Small delay to ensure navigation completes
|
|
await asyncio.sleep(1)
|
|
|
|
# Get browser state after navigation
|
|
current_url = await browser_session.get_current_page_url()
|
|
|
|
# Verify a new tab was created
|
|
final_tabs = await browser_session.get_tabs()
|
|
assert len(final_tabs) == initial_tabs_count + 1
|
|
|
|
# Verify focus switched to the new tab
|
|
assert 'example.com' in current_url
|
|
assert current_url != initial_url
|
|
|
|
async def test_navigate_and_verify_page_properties(self, browser_session, base_url):
|
|
"""Test that NavigateToUrlEvent changes the URL and page properties are accessible."""
|
|
from browser_use.browser.events import NavigateToUrlEvent
|
|
|
|
# Navigate to the test page
|
|
event = browser_session.event_bus.dispatch(NavigateToUrlEvent(url=f'{base_url}/'))
|
|
await event
|
|
|
|
# Wait for navigation to complete
|
|
await asyncio.sleep(0.5)
|
|
|
|
# Get the current page URL
|
|
current_url = await browser_session.get_current_page_url()
|
|
|
|
# Verify the page URL matches what we navigated to
|
|
assert f'{base_url}/' in current_url
|
|
|
|
# Verify the page title using the new API
|
|
title = await browser_session.get_current_page_title()
|
|
assert title == 'Test Home Page'
|