browser-use/tests/ci/test_sync_client.py

"""Tests for CloudSync client machinery - retry logic, event handling, backend communication."""

import os
import tempfile
from pathlib import Path

import httpx
import pytest
from bubus import BaseEvent
from pytest_httpserver import HTTPServer

from browser_use.agent.cloud_events import CreateAgentTaskEvent
from browser_use.sync.auth import TEMP_USER_ID, DeviceAuthClient
from browser_use.sync.service import CloudSync


@pytest.fixture
def temp_config_dir():
	"""Create temporary config directory for tests."""
	with tempfile.TemporaryDirectory() as tmpdir:
		temp_dir = Path(tmpdir) / '.config' / 'browseruse'
		temp_dir.mkdir(parents=True, exist_ok=True)

		os.environ['BROWSER_USE_CONFIG_DIR'] = str(temp_dir)

		yield temp_dir


@pytest.fixture
async def http_client(httpserver: HTTPServer):
	"""Create a real HTTP client pointed at the test server."""
	async with httpx.AsyncClient(base_url=httpserver.url_for('')) as client:
		yield client


class TestCloudSyncInit:
	"""Test CloudSync initialization and configuration."""

	async def test_init_with_auth_enabled(self, temp_config_dir):
		"""Test CloudSync initialization with auth enabled."""
		service = CloudSync(enable_auth=True, base_url='http://localhost:8000')

		assert service.base_url == 'http://localhost:8000'
		assert service.enable_auth is True
		assert service.auth_client is not None
		assert isinstance(service.auth_client, DeviceAuthClient)
		assert service.pending_events == []
		assert service.session_id is None

	async def test_init_with_auth_disabled(self, temp_config_dir):
		"""Test CloudSync initialization with auth disabled."""
		service = CloudSync(enable_auth=False, base_url='http://localhost:8000')

		assert service.base_url == 'http://localhost:8000'
		assert service.enable_auth is False
		assert service.auth_client is None
		assert service.pending_events == []


class TestCloudSyncEventHandling:
	"""Test CloudSync event validation and processing."""

	@pytest.fixture
	def authenticated_sync(self, httpserver: HTTPServer, http_client, temp_config_dir):
		"""Create authenticated CloudSync service."""
		auth = DeviceAuthClient(base_url=httpserver.url_for(''), http_client=http_client)
		auth.auth_config.api_token = 'test-api-key'
		auth.auth_config.user_id = 'test-user-123'

		service = CloudSync(base_url=httpserver.url_for(''), enable_auth=True)
		service.auth_client = auth
		service.session_id = 'test-session-id'
		return service

	@pytest.fixture
	def unauthenticated_sync(self, httpserver: HTTPServer, temp_config_dir):
		"""Create unauthenticated CloudSync service."""
		service = CloudSync(base_url=httpserver.url_for(''), enable_auth=True)
		service.session_id = 'test-session-id'
		return service

	async def test_event_forwarding_authenticated(self, httpserver: HTTPServer, authenticated_sync):
		"""Test event forwarding when authenticated."""
		# Capture requests
		requests = []

		def capture_request(request):
			requests.append(request.get_json())
			from werkzeug.wrappers import Response

			return Response('{"processed": 1, "failed": 0}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(capture_request)

		# Send event
		await authenticated_sync.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Test task',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Verify forwarding
		assert len(requests) == 1
		event_batch = requests[0]
		assert len(event_batch['events']) == 1

		event = event_batch['events'][0]
		assert event['event_type'] == 'CreateAgentTaskEvent'
		assert event['user_id'] == 'test-user-123'
		# BaseEvent creates event_type attribute, plus our custom data as attributes
		assert event['task'] == 'Test task'

	async def test_event_queueing_unauthenticated(self, httpserver: HTTPServer, unauthenticated_sync):
		"""Test event queueing when unauthenticated."""
		# Server returns 401
		httpserver.expect_request('/api/v1/events', method='POST').respond_with_json({'error': 'unauthorized'}, status=401)

		# Send event
		await unauthenticated_sync.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Queued task',
				user_id=TEMP_USER_ID,
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Event should be queued
		assert len(unauthenticated_sync.pending_events) == 1
		queued_event = unauthenticated_sync.pending_events[0]
		assert queued_event.event_type == 'CreateAgentTaskEvent'
		assert queued_event.user_id == TEMP_USER_ID
		assert queued_event.task == 'Queued task'

	async def test_event_user_id_injection_pre_auth(self, httpserver: HTTPServer, unauthenticated_sync):
		"""Test that temp user ID is injected for pre-auth events."""
		requests = []

		def capture_request(request):
			requests.append(request.get_json())
			from werkzeug.wrappers import Response

			return Response('{"processed": 1, "failed": 0}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(capture_request)

		# Send event without user_id
		await unauthenticated_sync.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Pre-auth task',
				user_id=TEMP_USER_ID,
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Verify temp user ID was injected
		assert len(requests) == 1
		event = requests[0]['events'][0]
		assert event['user_id'] == TEMP_USER_ID


class TestCloudSyncRetryLogic:
	"""Test CloudSync retry and error handling logic."""

	@pytest.fixture
	def sync_with_auth(self, httpserver: HTTPServer, http_client, temp_config_dir):
		"""Create CloudSync with auth."""
		auth = DeviceAuthClient(base_url=httpserver.url_for(''), http_client=http_client)
		auth.auth_config.api_token = 'test-api-key'
		auth.auth_config.user_id = 'test-user-123'

		service = CloudSync(base_url=httpserver.url_for(''), enable_auth=True)
		service.auth_client = auth
		service.session_id = 'test-session-id'
		return service

	async def test_pending_event_resending(self, httpserver: HTTPServer, sync_with_auth):
		"""Test resending of pending events after authentication."""
		requests = []

		def capture_request(request):
			requests.append(request.get_json())
			from werkzeug.wrappers import Response

			return Response('{"processed": 1, "failed": 0}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(capture_request)

		# Manually add pending events (simulating 401 scenario)
		sync_with_auth.pending_events.extend(
			[
				CreateAgentTaskEvent(
					agent_session_id='test-session',
					llm_model='test-model',
					task='Pending task 1',
					user_id=TEMP_USER_ID,
					device_id='test-device-id',
					done_output=None,
					user_feedback_type=None,
					user_comment=None,
					gif_url=None,
				),
				CreateAgentTaskEvent(
					agent_session_id='test-session',
					llm_model='test-model',
					task='Pending task 2',
					user_id=TEMP_USER_ID,
					device_id='test-device-id',
					done_output=None,
					user_feedback_type=None,
					user_comment=None,
					gif_url=None,
				),
			]
		)

		# Resend pending events
		await sync_with_auth._resend_pending_events()

		# Should send all pending events with updated user ID
		assert len(requests) == 2
		for i, request in enumerate(requests):
			event = request['events'][0]
			assert event['user_id'] == 'test-user-123'  # Updated from temp ID
			assert f'Pending task {i + 1}' == event['task']

		# Pending events should be cleared
		assert len(sync_with_auth.pending_events) == 0

	async def test_backend_error_resilience(self, httpserver: HTTPServer, sync_with_auth):
		"""Test resilience to backend errors."""
		# Server returns 500 error
		httpserver.expect_request('/api/v1/events', method='POST').respond_with_data('Internal Server Error', status=500)

		# Should not raise exception
		await sync_with_auth.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Task during outage',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Events should not be queued for 500 errors (only 401)
		assert len(sync_with_auth.pending_events) == 0

	async def test_network_error_resilience(self, sync_with_auth):
		"""Test resilience to network errors."""
		# No server running - will get connection error
		sync_with_auth.base_url = 'http://localhost:99999'  # Invalid port

		# Should not raise exception
		await sync_with_auth.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Task during network error',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Should handle gracefully without crashing

	async def test_concurrent_event_sending(self, httpserver: HTTPServer, sync_with_auth):
		"""Test handling of concurrent event sending."""
		import asyncio

		requests = []

		def capture_request(request):
			requests.append(request.get_json())
			from werkzeug.wrappers import Response

			return Response('{"processed": 1, "failed": 0}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(capture_request)

		# Send multiple events concurrently
		tasks = []
		for i in range(5):
			task = sync_with_auth.handle_event(
				CreateAgentTaskEvent(
					agent_session_id='test-session',
					llm_model='test-model',
					task=f'Concurrent task {i}',
					user_id='test-user-123',
					device_id='test-device-id',
					done_output=None,
					user_feedback_type=None,
					user_comment=None,
					gif_url=None,
				)
			)
			tasks.append(task)

		await asyncio.gather(*tasks)

		# All events should be sent
		assert len(requests) == 5
		# Just verify all events have task data - order may vary due to concurrency
		task_values = [req['events'][0]['task'] for req in requests]
		expected_tasks = [f'Concurrent task {i}' for i in range(5)]
		assert sorted(task_values) == sorted(expected_tasks)


class TestCloudSyncBackendCommunication:
	"""Test CloudSync backend communication patterns."""

	async def test_request_format_validation(self, httpserver: HTTPServer, temp_config_dir):
		"""Test that requests are formatted correctly for backend."""
		requests = []

		def capture_request(request):
			# Validate request structure
			assert request.content_type == 'application/json'
			data = request.get_json()
			requests.append(data)

			# Validate batch structure
			assert 'events' in data
			assert isinstance(data['events'], list)
			assert len(data['events']) == 1

			event = data['events'][0]
			required_fields = ['event_type', 'event_id', 'event_created_at', 'event_schema', 'user_id']
			for field in required_fields:
				assert field in event, f'Missing required field: {field}'

			from werkzeug.wrappers import Response

			return Response('{"processed": 1, "failed": 0}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(capture_request)

		# Create authenticated service
		auth = DeviceAuthClient(base_url=httpserver.url_for(''))
		auth.auth_config.api_token = 'test-api-key'
		auth.auth_config.user_id = 'test-user-123'

		service = CloudSync(base_url=httpserver.url_for(''), enable_auth=True)
		service.auth_client = auth
		service.session_id = 'test-session-id'

		await service.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Format validation test',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		assert len(requests) == 1

	async def test_auth_header_handling(self, httpserver: HTTPServer, temp_config_dir):
		"""Test proper auth header handling."""
		requests = []

		def capture_request(request):
			requests.append(
				{
					'headers': dict(request.headers),
					'json': request.get_json(),
				}
			)
			from werkzeug.wrappers import Response

			return Response('{"processed": 1, "failed": 0}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(capture_request)

		# Test authenticated request
		auth = DeviceAuthClient(base_url=httpserver.url_for(''))
		auth.auth_config.api_token = 'test-api-key'
		auth.auth_config.user_id = 'test-user-123'

		service = CloudSync(base_url=httpserver.url_for(''), enable_auth=True)
		service.auth_client = auth

		await service.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Auth header test',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Check auth header was included
		assert len(requests) == 1
		headers = requests[0]['headers']
		assert 'Authorization' in headers
		assert headers['Authorization'] == 'Bearer test-api-key'

		# Test unauthenticated request
		requests.clear()
		service.auth_client = DeviceAuthClient(base_url=httpserver.url_for(''))  # No credentials

		await service.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='No auth test',
				user_id='',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

		# Check no auth header
		assert len(requests) == 1
		headers = requests[0]['headers']
		assert 'Authorization' not in headers


class TestCloudSyncErrorHandling:
	"""Test CloudSync error handling doesn't crash the agent."""

	@pytest.fixture
	def sync_service(self, httpserver: HTTPServer, temp_config_dir):
		"""Create CloudSync service."""
		return CloudSync(base_url=httpserver.url_for(''), enable_auth=False)

	async def test_timeout_error_handling(self, sync_service):
		"""Test that timeout errors are handled gracefully."""
		# Use a URL that will timeout
		sync_service.base_url = 'http://10.255.255.1'  # Non-routable IP for timeout

		# Should not raise exception
		await sync_service.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Timeout test',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

	async def test_malformed_event_handling(self, httpserver: HTTPServer, sync_service):
		"""Test handling of events that can't be serialized."""

		class BadEvent(BaseEvent):
			"""Event that will fail to serialize."""

			event_type: str = 'BadEvent'

			def model_dump(self, **kwargs):
				raise ValueError('Serialization failed')

		# Should not raise exception
		await sync_service.handle_event(BadEvent())

	async def test_http_error_responses(self, httpserver: HTTPServer, sync_service):
		"""Test various HTTP error responses don't crash the service."""
		error_codes = [400, 403, 404, 429, 500, 502, 503]

		for status_code in error_codes:
			httpserver.expect_request('/api/v1/events', method='POST').respond_with_json(
				{'error': f'Test error {status_code}'}, status=status_code
			)

			# Should not raise exception
			await sync_service.handle_event(
				CreateAgentTaskEvent(
					agent_session_id='test-session',
					llm_model='test-model',
					task=f'Error {status_code} test',
					user_id='test-user-123',
					device_id='test-device-id',
					done_output=None,
					user_feedback_type=None,
					user_comment=None,
					gif_url=None,
				)
			)

	async def test_invalid_response_handling(self, httpserver: HTTPServer, sync_service):
		"""Test handling of invalid server responses."""
		# Return invalid JSON
		httpserver.expect_request('/api/v1/events', method='POST').respond_with_data('Not JSON', status=200)

		# Should not raise exception
		await sync_service.handle_event(
			CreateAgentTaskEvent(
				agent_session_id='test-session',
				llm_model='test-model',
				task='Invalid response test',
				user_id='test-user-123',
				device_id='test-device-id',
				done_output=None,
				user_feedback_type=None,
				user_comment=None,
				gif_url=None,
			)
		)

	async def test_event_with_restricted_attributes(self, httpserver: HTTPServer, sync_service):
		"""Test handling events that don't allow user_id attribute."""
		from pydantic import ConfigDict

		class RestrictedEvent(BaseEvent):
			"""Event that doesn't allow extra attributes."""

			model_config = ConfigDict(extra='forbid')
			event_type: str = 'RestrictedEvent'
			data: str = 'test'

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_json({'processed': 1}, status=200)

		# Should not raise exception - will log debug message about not being able to set user_id
		await sync_service.handle_event(RestrictedEvent())

	async def test_concurrent_error_resilience(self, httpserver: HTTPServer, sync_service):
		"""Test that concurrent errors don't affect other events."""
		import asyncio

		successful_requests = []
		request_count = 0

		def handler(request):
			nonlocal request_count
			request_count += 1
			# Every 3rd request fails
			if request_count % 3 == 0:
				from werkzeug.wrappers import Response

				return Response('Server Error', status=500)
			else:
				successful_requests.append(request.get_json())
				from werkzeug.wrappers import Response

				return Response('{"processed": 1}', status=200, mimetype='application/json')

		httpserver.expect_request('/api/v1/events', method='POST').respond_with_handler(handler)

		# Send 10 events concurrently
		tasks = []
		for i in range(10):
			task = sync_service.handle_event(
				CreateAgentTaskEvent(
					agent_session_id='test-session',
					llm_model='test-model',
					task=f'Concurrent error test {i}',
					user_id='test-user-123',
					device_id='test-device-id',
					done_output=None,
					user_feedback_type=None,
					user_comment=None,
					gif_url=None,
				)
			)
			tasks.append(task)

		# All should complete without raising
		await asyncio.gather(*tasks)

		# ~7 should succeed (10 total, ~3 fail)
		assert len(successful_requests) >= 6