Spaces:
Sleeping
Sleeping
File size: 8,891 Bytes
d7b3d84 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 |
"""Tests for cloud browser functionality."""
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from browser_use.browser.cloud.cloud import (
CloudBrowserAuthError,
CloudBrowserClient,
CloudBrowserError,
)
from browser_use.browser.cloud.views import CreateBrowserRequest
from browser_use.browser.profile import BrowserProfile
from browser_use.browser.session import BrowserSession
from browser_use.sync.auth import CloudAuthConfig
@pytest.fixture
def temp_config_dir(monkeypatch):
"""Create temporary config directory."""
with tempfile.TemporaryDirectory() as tmpdir:
temp_dir = Path(tmpdir) / '.config' / 'browseruse'
temp_dir.mkdir(parents=True, exist_ok=True)
# Use monkeypatch to set the environment variable
monkeypatch.setenv('BROWSER_USE_CONFIG_DIR', str(temp_dir))
yield temp_dir
@pytest.fixture
def mock_auth_config(temp_config_dir):
"""Create a mock auth config with valid token."""
auth_config = CloudAuthConfig(api_token='test-token', user_id='test-user-id', authorized_at=None)
auth_config.save_to_file()
return auth_config
class TestCloudBrowserClient:
"""Test CloudBrowserClient class."""
async def test_create_browser_success(self, mock_auth_config, monkeypatch):
"""Test successful cloud browser creation."""
# Clear environment variable so test uses mock_auth_config
monkeypatch.delenv('BROWSER_USE_API_KEY', raising=False)
# Mock response data matching the API
mock_response_data = {
'id': 'test-browser-id',
'status': 'active',
'liveUrl': 'https://live.browser-use.com?wss=test',
'cdpUrl': 'wss://test.proxy.daytona.works',
'timeoutAt': '2025-09-17T04:35:36.049892',
'startedAt': '2025-09-17T03:35:36.049974',
'finishedAt': None,
}
# Mock the httpx client
with patch('httpx.AsyncClient') as mock_client_class:
mock_response = AsyncMock()
mock_response.status_code = 201
mock_response.is_success = True
mock_response.json = lambda: mock_response_data
mock_client = AsyncMock()
mock_client.post.return_value = mock_response
mock_client_class.return_value = mock_client
client = CloudBrowserClient()
client.client = mock_client
result = await client.create_browser(CreateBrowserRequest())
assert result.id == 'test-browser-id'
assert result.status == 'active'
assert result.cdpUrl == 'wss://test.proxy.daytona.works'
# Verify auth headers were included
mock_client.post.assert_called_once()
call_args = mock_client.post.call_args
assert 'X-Browser-Use-API-Key' in call_args.kwargs['headers']
assert call_args.kwargs['headers']['X-Browser-Use-API-Key'] == 'test-token'
async def test_create_browser_auth_error(self, temp_config_dir, monkeypatch):
"""Test cloud browser creation with auth error."""
# Clear environment variable and don't create auth config - should trigger auth error
monkeypatch.delenv('BROWSER_USE_API_KEY', raising=False)
client = CloudBrowserClient()
with pytest.raises(CloudBrowserAuthError) as exc_info:
await client.create_browser(CreateBrowserRequest())
assert 'BROWSER_USE_API_KEY environment variable' in str(exc_info.value)
async def test_create_browser_http_401(self, mock_auth_config, monkeypatch):
"""Test cloud browser creation with HTTP 401 response."""
# Clear environment variable so test uses mock_auth_config
monkeypatch.delenv('BROWSER_USE_API_KEY', raising=False)
with patch('httpx.AsyncClient') as mock_client_class:
mock_response = AsyncMock()
mock_response.status_code = 401
mock_response.is_success = False
mock_client = AsyncMock()
mock_client.post.return_value = mock_response
mock_client_class.return_value = mock_client
client = CloudBrowserClient()
client.client = mock_client
with pytest.raises(CloudBrowserAuthError) as exc_info:
await client.create_browser(CreateBrowserRequest())
assert 'Authentication failed' in str(exc_info.value)
async def test_create_browser_with_env_var(self, temp_config_dir, monkeypatch):
"""Test cloud browser creation using BROWSER_USE_API_KEY environment variable."""
# Set environment variable
monkeypatch.setenv('BROWSER_USE_API_KEY', 'env-test-token')
# Mock response data matching the API
mock_response_data = {
'id': 'test-browser-id',
'status': 'active',
'liveUrl': 'https://live.browser-use.com?wss=test',
'cdpUrl': 'wss://test.proxy.daytona.works',
'timeoutAt': '2025-09-17T04:35:36.049892',
'startedAt': '2025-09-17T03:35:36.049974',
'finishedAt': None,
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_response = AsyncMock()
mock_response.status_code = 201
mock_response.is_success = True
mock_response.json = lambda: mock_response_data
mock_client = AsyncMock()
mock_client.post.return_value = mock_response
mock_client_class.return_value = mock_client
client = CloudBrowserClient()
client.client = mock_client
result = await client.create_browser(CreateBrowserRequest())
assert result.id == 'test-browser-id'
assert result.status == 'active'
assert result.cdpUrl == 'wss://test.proxy.daytona.works'
# Verify environment variable was used
mock_client.post.assert_called_once()
call_args = mock_client.post.call_args
assert 'X-Browser-Use-API-Key' in call_args.kwargs['headers']
assert call_args.kwargs['headers']['X-Browser-Use-API-Key'] == 'env-test-token'
async def test_stop_browser_success(self, mock_auth_config, monkeypatch):
"""Test successful cloud browser session stop."""
# Clear environment variable so test uses mock_auth_config
monkeypatch.delenv('BROWSER_USE_API_KEY', raising=False)
# Mock response data for stop
mock_response_data = {
'id': 'test-browser-id',
'status': 'stopped',
'liveUrl': 'https://live.browser-use.com?wss=test',
'cdpUrl': 'wss://test.proxy.daytona.works',
'timeoutAt': '2025-09-17T04:35:36.049892',
'startedAt': '2025-09-17T03:35:36.049974',
'finishedAt': '2025-09-17T04:35:36.049892',
}
with patch('httpx.AsyncClient') as mock_client_class:
mock_response = AsyncMock()
mock_response.status_code = 200
mock_response.is_success = True
mock_response.json = lambda: mock_response_data
mock_client = AsyncMock()
mock_client.patch.return_value = mock_response
mock_client_class.return_value = mock_client
client = CloudBrowserClient()
client.client = mock_client
client.current_session_id = 'test-browser-id'
result = await client.stop_browser()
assert result.id == 'test-browser-id'
assert result.status == 'stopped'
assert result.finishedAt is not None
# Verify correct API call
mock_client.patch.assert_called_once()
call_args = mock_client.patch.call_args
assert 'test-browser-id' in call_args.args[0] # URL contains session ID
assert call_args.kwargs['json'] == {'action': 'stop'}
assert 'X-Browser-Use-API-Key' in call_args.kwargs['headers']
async def test_stop_browser_session_not_found(self, mock_auth_config, monkeypatch):
"""Test stopping a browser session that doesn't exist."""
# Clear environment variable so test uses mock_auth_config
monkeypatch.delenv('BROWSER_USE_API_KEY', raising=False)
with patch('httpx.AsyncClient') as mock_client_class:
mock_response = AsyncMock()
mock_response.status_code = 404
mock_response.is_success = False
mock_client = AsyncMock()
mock_client.patch.return_value = mock_response
mock_client_class.return_value = mock_client
client = CloudBrowserClient()
client.client = mock_client
with pytest.raises(CloudBrowserError) as exc_info:
await client.stop_browser('nonexistent-session')
assert 'not found' in str(exc_info.value)
class TestBrowserSessionCloudIntegration:
"""Test BrowserSession integration with cloud browsers."""
async def test_cloud_browser_profile_property(self):
"""Test that cloud_browser property works correctly."""
# Just test the profile and session properties without connecting
profile = BrowserProfile(use_cloud=True)
session = BrowserSession(browser_profile=profile, cdp_url='ws://mock-url') # Provide CDP URL to avoid connection
assert session.cloud_browser is True
assert session.browser_profile.use_cloud is True
async def test_browser_session_cloud_browser_logic(self, mock_auth_config, monkeypatch):
"""Test that cloud browser profile settings work correctly."""
# Clear environment variable so test uses mock_auth_config
monkeypatch.delenv('BROWSER_USE_API_KEY', raising=False)
# Test cloud browser profile creation
profile = BrowserProfile(use_cloud=True)
assert profile.use_cloud is True
# Test that BrowserSession respects cloud_browser setting
# Provide CDP URL to avoid actual connection attempts
session = BrowserSession(browser_profile=profile, cdp_url='ws://mock-url')
assert session.cloud_browser is True
|