"""Unit tests for git and security functionality in AppConversationServiceBase. This module tests the git-related functionality, specifically the clone_or_init_git_repo method and the recent bug fixes for git checkout operations. """ import subprocess from pathlib import Path from types import MethodType from unittest.mock import AsyncMock, MagicMock, Mock, patch from uuid import uuid4 import pytest from openhands.app_server.app_conversation.app_conversation_models import AgentType from openhands.app_server.app_conversation.app_conversation_service_base import ( AppConversationServiceBase, ) from openhands.app_server.integrations.service_types import ProviderType from openhands.app_server.sandbox.sandbox_models import SandboxInfo, SandboxStatus from openhands.app_server.user.user_context import UserContext from openhands.sdk.skills import Skill class MockUserInfo: """Mock class for UserInfo to simulate user settings.""" def __init__( self, git_user_name: str | None = None, git_user_email: str | None = None ): self.git_user_name = git_user_name self.git_user_email = git_user_email class MockCommandResult: """Mock class for command execution result.""" def __init__(self, exit_code: int = 0, stderr: str = ''): self.exit_code = exit_code self.stderr = stderr class MockWorkspace: """Mock class for AsyncRemoteWorkspace.""" def __init__(self, working_dir: str = '/workspace'): self.working_dir = working_dir self.execute_command = AsyncMock(return_value=MockCommandResult()) class MockAppConversationServiceBase: """Mock class to test git functionality without complex dependencies.""" def __init__(self): self.logger = MagicMock() async def clone_or_init_git_repo( self, workspace_path: str, repo_url: str, branch: str = 'main', timeout: int = 300, ) -> bool: """Clone or initialize a git repository. This is a simplified version of the actual method for testing purposes. """ try: # Try to clone the repository clone_result = subprocess.run( ['git', 'clone', '--branch', branch, repo_url, workspace_path], capture_output=True, text=True, timeout=timeout, ) if clone_result.returncode == 0: self.logger.info( f'Successfully cloned repository {repo_url} to {workspace_path}' ) return True # If clone fails, try to checkout the branch checkout_result = subprocess.run( ['git', 'checkout', branch], cwd=workspace_path, capture_output=True, text=True, timeout=timeout, ) if checkout_result.returncode == 0: self.logger.info(f'Successfully checked out branch {branch}') return True else: self.logger.error( f'Failed to checkout branch {branch}: {checkout_result.stderr}' ) return False except subprocess.TimeoutExpired: self.logger.error(f'Git operation timed out after {timeout} seconds') return False except Exception as e: self.logger.error(f'Git operation failed: {str(e)}') return False @pytest.fixture def service(): """Create a mock service instance for testing.""" return MockAppConversationServiceBase() @pytest.mark.asyncio async def test_clone_or_init_git_repo_successful_clone(service): """Test successful git clone operation.""" with patch('subprocess.run') as mock_run: # Mock successful clone mock_run.return_value = MagicMock(returncode=0, stderr='', stdout='Cloning...') result = await service.clone_or_init_git_repo( workspace_path='/tmp/test_repo', repo_url='https://github.com/test/repo.git', branch='main', timeout=300, ) assert result is True mock_run.assert_called_once_with( [ 'git', 'clone', '--branch', 'main', 'https://github.com/test/repo.git', '/tmp/test_repo', ], capture_output=True, text=True, timeout=300, ) service.logger.info.assert_called_with( 'Successfully cloned repository https://github.com/test/repo.git to /tmp/test_repo' ) @pytest.mark.asyncio async def test_clone_or_init_git_repo_clone_fails_checkout_succeeds(service): """Test git clone fails but checkout succeeds.""" with patch('subprocess.run') as mock_run: # Mock clone failure, then checkout success mock_run.side_effect = [ MagicMock(returncode=1, stderr='Clone failed', stdout=''), # Clone fails MagicMock( returncode=0, stderr='', stdout='Switched to branch' ), # Checkout succeeds ] result = await service.clone_or_init_git_repo( workspace_path='/tmp/test_repo', repo_url='https://github.com/test/repo.git', branch='feature-branch', timeout=300, ) assert result is True assert mock_run.call_count == 2 # Check clone call mock_run.assert_any_call( [ 'git', 'clone', '--branch', 'feature-branch', 'https://github.com/test/repo.git', '/tmp/test_repo', ], capture_output=True, text=True, timeout=300, ) # Check checkout call mock_run.assert_any_call( ['git', 'checkout', 'feature-branch'], cwd='/tmp/test_repo', capture_output=True, text=True, timeout=300, ) service.logger.info.assert_called_with( 'Successfully checked out branch feature-branch' ) @pytest.mark.asyncio async def test_clone_or_init_git_repo_both_operations_fail(service): """Test both git clone and checkout operations fail.""" with patch('subprocess.run') as mock_run: # Mock both operations failing mock_run.side_effect = [ MagicMock(returncode=1, stderr='Clone failed', stdout=''), # Clone fails MagicMock( returncode=1, stderr='Checkout failed', stdout='' ), # Checkout fails ] result = await service.clone_or_init_git_repo( workspace_path='/tmp/test_repo', repo_url='https://github.com/test/repo.git', branch='nonexistent-branch', timeout=300, ) assert result is False assert mock_run.call_count == 2 service.logger.error.assert_called_with( 'Failed to checkout branch nonexistent-branch: Checkout failed' ) @pytest.mark.asyncio async def test_clone_or_init_git_repo_timeout(service): """Test git operation timeout.""" with patch('subprocess.run') as mock_run: # Mock timeout exception mock_run.side_effect = subprocess.TimeoutExpired( cmd=['git', 'clone'], timeout=300 ) result = await service.clone_or_init_git_repo( workspace_path='/tmp/test_repo', repo_url='https://github.com/test/repo.git', branch='main', timeout=300, ) assert result is False service.logger.error.assert_called_with( 'Git operation timed out after 300 seconds' ) @pytest.mark.asyncio async def test_clone_or_init_git_repo_exception(service): """Test git operation with unexpected exception.""" with patch('subprocess.run') as mock_run: # Mock unexpected exception mock_run.side_effect = Exception('Unexpected error') result = await service.clone_or_init_git_repo( workspace_path='/tmp/test_repo', repo_url='https://github.com/test/repo.git', branch='main', timeout=300, ) assert result is False service.logger.error.assert_called_with( 'Git operation failed: Unexpected error' ) @pytest.mark.asyncio async def test_clone_or_init_git_repo_custom_timeout(service): """Test git operation with custom timeout.""" with patch('subprocess.run') as mock_run: # Mock successful clone with custom timeout mock_run.return_value = MagicMock(returncode=0, stderr='', stdout='Cloning...') result = await service.clone_or_init_git_repo( workspace_path='/tmp/test_repo', repo_url='https://github.com/test/repo.git', branch='main', timeout=600, # Custom timeout ) assert result is True mock_run.assert_called_once_with( [ 'git', 'clone', '--branch', 'main', 'https://github.com/test/repo.git', '/tmp/test_repo', ], capture_output=True, text=True, timeout=600, # Verify custom timeout is used ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.LLMSummarizingCondenser' ) def test_create_condenser_default_agent_with_none_max_size(mock_condenser_class): """Test _create_condenser for DEFAULT agent with condenser_max_size = None uses default.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object( AppConversationServiceBase, '__abstractmethods__', set(), ): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context, ) mock_llm = MagicMock() mock_llm_copy = MagicMock() mock_llm_copy.usage_id = 'condenser' mock_llm.model_copy.return_value = mock_llm_copy mock_condenser_instance = MagicMock() mock_condenser_class.return_value = mock_condenser_instance # Act service._create_condenser(mock_llm, AgentType.DEFAULT, None) # Assert mock_condenser_class.assert_called_once() call_kwargs = mock_condenser_class.call_args[1] # When condenser_max_size is None, max_size should not be passed (uses SDK default of 240) assert 'max_size' not in call_kwargs # keep_first is never passed (uses SDK default of 2) assert 'keep_first' not in call_kwargs assert call_kwargs['llm'].usage_id == 'condenser' mock_llm.model_copy.assert_called_once() @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.LLMSummarizingCondenser' ) def test_create_condenser_default_agent_with_custom_max_size(mock_condenser_class): """Test _create_condenser for DEFAULT agent with custom condenser_max_size.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object( AppConversationServiceBase, '__abstractmethods__', set(), ): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context, ) mock_llm = MagicMock() mock_llm_copy = MagicMock() mock_llm_copy.usage_id = 'condenser' mock_llm.model_copy.return_value = mock_llm_copy mock_condenser_instance = MagicMock() mock_condenser_class.return_value = mock_condenser_instance # Act service._create_condenser(mock_llm, AgentType.DEFAULT, 150) # Assert mock_condenser_class.assert_called_once() call_kwargs = mock_condenser_class.call_args[1] assert call_kwargs['max_size'] == 150 # Custom value should be used # keep_first is never passed (uses SDK default of 2) assert 'keep_first' not in call_kwargs assert call_kwargs['llm'].usage_id == 'condenser' mock_llm.model_copy.assert_called_once() @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.LLMSummarizingCondenser' ) def test_create_condenser_plan_agent_with_none_max_size(mock_condenser_class): """Test _create_condenser for PLAN agent with condenser_max_size = None uses default.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object( AppConversationServiceBase, '__abstractmethods__', set(), ): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context, ) mock_llm = MagicMock() mock_llm_copy = MagicMock() mock_llm_copy.usage_id = 'planning_condenser' mock_llm.model_copy.return_value = mock_llm_copy mock_condenser_instance = MagicMock() mock_condenser_class.return_value = mock_condenser_instance # Act service._create_condenser(mock_llm, AgentType.PLAN, None) # Assert mock_condenser_class.assert_called_once() call_kwargs = mock_condenser_class.call_args[1] # When condenser_max_size is None, max_size should not be passed (uses SDK default of 240) assert 'max_size' not in call_kwargs # keep_first is never passed (uses SDK default of 2) assert 'keep_first' not in call_kwargs assert call_kwargs['llm'].usage_id == 'planning_condenser' mock_llm.model_copy.assert_called_once() @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.LLMSummarizingCondenser' ) def test_create_condenser_plan_agent_with_custom_max_size(mock_condenser_class): """Test _create_condenser for PLAN agent with custom condenser_max_size.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object( AppConversationServiceBase, '__abstractmethods__', set(), ): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context, ) mock_llm = MagicMock() mock_llm_copy = MagicMock() mock_llm_copy.usage_id = 'planning_condenser' mock_llm.model_copy.return_value = mock_llm_copy mock_condenser_instance = MagicMock() mock_condenser_class.return_value = mock_condenser_instance # Act service._create_condenser(mock_llm, AgentType.PLAN, 200) # Assert mock_condenser_class.assert_called_once() call_kwargs = mock_condenser_class.call_args[1] assert call_kwargs['max_size'] == 200 # Custom value should be used # keep_first is never passed (uses SDK default of 2) assert 'keep_first' not in call_kwargs assert call_kwargs['llm'].usage_id == 'planning_condenser' mock_llm.model_copy.assert_called_once() # ============================================================================= # Tests for security analyzer helpers # ============================================================================= @pytest.mark.parametrize('value', [None, '', 'none', 'NoNe']) def test_create_security_analyzer_returns_none_for_empty_values(value): """_create_security_analyzer_from_string returns None for empty/none values.""" # Arrange service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=('_create_security_analyzer_from_string',) ) # Act result = service._create_security_analyzer_from_string(value) # Assert assert result is None def test_create_security_analyzer_returns_llm_analyzer(): """_create_security_analyzer_from_string returns LLMSecurityAnalyzer for llm string.""" # Arrange security_analyzer_str = 'llm' service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=('_create_security_analyzer_from_string',) ) # Act result = service._create_security_analyzer_from_string(security_analyzer_str) # Assert from openhands.sdk.security import LLMSecurityAnalyzer assert isinstance(result, LLMSecurityAnalyzer) def test_create_security_analyzer_logs_warning_for_unknown_value(): """_create_security_analyzer_from_string logs warning and returns None for unknown.""" # Arrange unknown_value = 'custom' service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=('_create_security_analyzer_from_string',) ) # Act with patch( 'openhands.app_server.app_conversation.app_conversation_service_base._logger' ) as mock_logger: result = service._create_security_analyzer_from_string(unknown_value) # Assert assert result is None mock_logger.warning.assert_called_once() def test_select_confirmation_policy_when_disabled_returns_never_confirm(): """_select_confirmation_policy returns NeverConfirm when confirmation_mode is False.""" # Arrange confirmation_mode = False security_analyzer = 'llm' service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=('_select_confirmation_policy',) ) # Act policy = service._select_confirmation_policy(confirmation_mode, security_analyzer) # Assert from openhands.sdk.security import NeverConfirm assert isinstance(policy, NeverConfirm) def test_select_confirmation_policy_llm_returns_confirm_risky(): """_select_confirmation_policy uses ConfirmRisky when analyzer is llm.""" # Arrange confirmation_mode = True security_analyzer = 'llm' service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=('_select_confirmation_policy',) ) # Act policy = service._select_confirmation_policy(confirmation_mode, security_analyzer) # Assert from openhands.sdk.security import ConfirmRisky assert isinstance(policy, ConfirmRisky) @pytest.mark.parametrize('security_analyzer', [None, '', 'none', 'custom']) def test_select_confirmation_policy_non_llm_returns_always_confirm( security_analyzer, ): """_select_confirmation_policy falls back to AlwaysConfirm for non-llm values.""" # Arrange confirmation_mode = True service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=('_select_confirmation_policy',) ) # Act policy = service._select_confirmation_policy(confirmation_mode, security_analyzer) # Assert from openhands.sdk.security import AlwaysConfirm assert isinstance(policy, AlwaysConfirm) @pytest.mark.asyncio async def test_set_security_analyzer_skips_when_no_session_key(): """_set_security_analyzer_from_settings exits early without session_api_key.""" # Arrange agent_server_url = 'https://agent.example.com' conversation_id = uuid4() httpx_client = AsyncMock() service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=( '_create_security_analyzer_from_string', '_set_security_analyzer_from_settings', ), ) with patch.object(service, '_create_security_analyzer_from_string') as mock_create: # Act await service._set_security_analyzer_from_settings( agent_server_url=agent_server_url, session_api_key=None, conversation_id=conversation_id, security_analyzer_str='llm', httpx_client=httpx_client, ) # Assert mock_create.assert_not_called() httpx_client.post.assert_not_called() @pytest.mark.asyncio async def test_set_security_analyzer_skips_when_analyzer_none(): """_set_security_analyzer_from_settings skips API call when analyzer resolves to None.""" # Arrange agent_server_url = 'https://agent.example.com' session_api_key = 'session-key' conversation_id = uuid4() httpx_client = AsyncMock() service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=( '_create_security_analyzer_from_string', '_set_security_analyzer_from_settings', ), ) with patch.object( service, '_create_security_analyzer_from_string', return_value=None ) as mock_create: # Act await service._set_security_analyzer_from_settings( agent_server_url=agent_server_url, session_api_key=session_api_key, conversation_id=conversation_id, security_analyzer_str='none', httpx_client=httpx_client, ) # Assert mock_create.assert_called_once_with('none') httpx_client.post.assert_not_called() class DummyAnalyzer: """Simple analyzer stub for testing model_dump contract.""" def __init__(self, payload: dict): self._payload = payload def model_dump(self) -> dict: return self._payload @pytest.mark.asyncio async def test_set_security_analyzer_successfully_calls_agent_server(): """_set_security_analyzer_from_settings posts analyzer payload when available.""" # Arrange agent_server_url = 'https://agent.example.com' session_api_key = 'session-key' conversation_id = uuid4() analyzer_payload = {'type': 'llm'} httpx_client = AsyncMock() http_response = MagicMock() http_response.raise_for_status = MagicMock() httpx_client.post.return_value = http_response service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=( '_create_security_analyzer_from_string', '_set_security_analyzer_from_settings', ), ) analyzer = DummyAnalyzer(analyzer_payload) with ( patch.object( service, '_create_security_analyzer_from_string', return_value=analyzer, ) as mock_create, patch( 'openhands.app_server.app_conversation.app_conversation_service_base._logger' ) as mock_logger, ): # Act await service._set_security_analyzer_from_settings( agent_server_url=agent_server_url, session_api_key=session_api_key, conversation_id=conversation_id, security_analyzer_str='llm', httpx_client=httpx_client, ) # Assert mock_create.assert_called_once_with('llm') httpx_client.post.assert_awaited_once_with( f'{agent_server_url}/api/conversations/{conversation_id}/security_analyzer', json={'security_analyzer': analyzer_payload}, headers={'X-Session-API-Key': session_api_key}, timeout=30.0, ) http_response.raise_for_status.assert_called_once() mock_logger.info.assert_called() @pytest.mark.asyncio async def test_set_security_analyzer_logs_warning_on_failure(): """_set_security_analyzer_from_settings warns but does not raise on errors.""" # Arrange agent_server_url = 'https://agent.example.com' session_api_key = 'session-key' conversation_id = uuid4() analyzer_payload = {'type': 'llm'} httpx_client = AsyncMock() httpx_client.post.side_effect = RuntimeError('network down') service, _ = _create_service_with_mock_user_context( MockUserInfo(), bind_methods=( '_create_security_analyzer_from_string', '_set_security_analyzer_from_settings', ), ) analyzer = DummyAnalyzer(analyzer_payload) with ( patch.object( service, '_create_security_analyzer_from_string', return_value=analyzer, ) as mock_create, patch( 'openhands.app_server.app_conversation.app_conversation_service_base._logger' ) as mock_logger, ): # Act await service._set_security_analyzer_from_settings( agent_server_url=agent_server_url, session_api_key=session_api_key, conversation_id=conversation_id, security_analyzer_str='llm', httpx_client=httpx_client, ) # Assert mock_create.assert_called_once_with('llm') httpx_client.post.assert_awaited_once() mock_logger.warning.assert_called() # ============================================================================= # Tests for _configure_git_user_settings # ============================================================================= def _create_service_with_mock_user_context( user_info: MockUserInfo, bind_methods: tuple[str, ...] | None = None ) -> tuple: """Create a mock service with selected real methods bound for testing. Uses MagicMock for the service but binds the real method for testing. Returns a tuple of (service, mock_user_context) for testing. """ mock_user_context = MagicMock() mock_user_context.get_user_info = AsyncMock(return_value=user_info) # Create a simple mock service and set required attribute service = MagicMock() service.user_context = mock_user_context methods_to_bind = ['_configure_git_user_settings'] if bind_methods: methods_to_bind.extend(bind_methods) # Remove potential duplicates while keeping order methods_to_bind = list(dict.fromkeys(methods_to_bind)) # Bind actual methods from the real class to test implementations directly for method_name in methods_to_bind: real_method = getattr(AppConversationServiceBase, method_name) setattr(service, method_name, MethodType(real_method, service)) return service, mock_user_context @pytest.fixture def mock_workspace(): """Create a mock workspace instance for testing.""" return MockWorkspace(working_dir='/workspace/project') @pytest.mark.asyncio async def test_clone_or_init_git_repo_quotes_selected_branch_before_checkout( mock_workspace, ): user_info = MockUserInfo() service, mock_user_context = _create_service_with_mock_user_context( user_info, bind_methods=( 'clone_or_init_git_repo', '_get_azure_devops_bearer_token_for_git', ), ) service.init_git_in_empty_workspace = True mock_user_context.get_authenticated_git_url = AsyncMock( return_value='https://github.com/owner/repo.git' ) task = Mock() task.request = Mock( selected_repository='owner/repo', selected_branch='feature>tmp', ) await service.clone_or_init_git_repo(task, mock_workspace) mock_workspace.execute_command.assert_any_call( "git checkout 'feature>tmp'", Path(mock_workspace.working_dir) / 'repo', ) @pytest.mark.asyncio async def test_clone_or_init_git_repo_configures_dynamic_azure_devops_helper( mock_workspace, ): user_info = MockUserInfo() service, mock_user_context = _create_service_with_mock_user_context( user_info, bind_methods=( 'clone_or_init_git_repo', '_get_azure_devops_bearer_token_for_git', '_configure_azure_devops_git_credential_helper', ), ) service.init_git_in_empty_workspace = True mock_user_context.get_authenticated_git_url = AsyncMock( return_value='https://dev.azure.com/org/project/_git/repo' ) mock_user_context.get_latest_token = AsyncMock( return_value='header.payload.signature' ) task = Mock() task.request = Mock( selected_repository='org/project/repo', selected_branch='main', git_provider=ProviderType.AZURE_DEVOPS, ) sandbox = SandboxInfo( id='sandbox-123', created_by_user_id='user-123', sandbox_spec_id='spec-123', status=SandboxStatus.RUNNING, session_api_key='session-key', ) await service.clone_or_init_git_repo(task, mock_workspace, sandbox) commands = [call.args[0] for call in mock_workspace.execute_command.call_args_list] assert any( "git -c http.extraheader='Authorization: Bearer header.payload.signature' clone" in command for command in commands ) helper_command = next( command for command in commands if 'openhands-azure-devops-credential-helper' in command ) assert ( '/api/v1/sandboxes/sandbox-123/settings/secrets/azure_devops_token' in helper_command ) assert ( 'git config --local --unset-all http.https://dev.azure.com/org/.extraheader' in helper_command ) assert 'credential.https://dev.azure.com/org.helper' in helper_command assert not any( command.startswith( 'git config --local http.https://dev.azure.com/org/.extraheader' ) for command in commands ) @pytest.mark.asyncio async def test_azure_devops_git_credential_helper_logs_without_web_url( mock_workspace, ): user_info = MockUserInfo() service, _ = _create_service_with_mock_user_context( user_info, bind_methods=('_configure_azure_devops_git_credential_helper',), ) service.web_url = None sandbox = SandboxInfo( id='sandbox-123', created_by_user_id='user-123', sandbox_spec_id='spec-123', status=SandboxStatus.RUNNING, session_api_key='session-key', ) with patch( 'openhands.app_server.app_conversation.app_conversation_service_base._logger.debug' ) as mock_debug: await service._configure_azure_devops_git_credential_helper( mock_workspace, Path(mock_workspace.working_dir), 'org/project/repo', sandbox, ) mock_debug.assert_called_once_with( 'Azure DevOps git credential helper has no configured web_url; ' 'it will rely on OH_WEBHOOKS_0_BASE_URL at runtime.' ) @pytest.mark.asyncio async def test_configure_git_user_settings_both_name_and_email(mock_workspace): """Test configuring both git user name and email.""" user_info = MockUserInfo( git_user_name='Test User', git_user_email='test@example.com' ) service, mock_user_context = _create_service_with_mock_user_context(user_info) await service._configure_git_user_settings(mock_workspace) # Verify get_user_info was called mock_user_context.get_user_info.assert_called_once() # Verify both git config commands were executed assert mock_workspace.execute_command.call_count == 2 # Check git config user.name call mock_workspace.execute_command.assert_any_call( 'git config --global user.name "Test User"', '/workspace/project' ) # Check git config user.email call mock_workspace.execute_command.assert_any_call( 'git config --global user.email "test@example.com"', '/workspace/project' ) @pytest.mark.asyncio async def test_configure_git_user_settings_only_name(mock_workspace): """Test configuring only git user name.""" user_info = MockUserInfo(git_user_name='Test User', git_user_email=None) service, _ = _create_service_with_mock_user_context(user_info) await service._configure_git_user_settings(mock_workspace) # Verify only user.name was configured assert mock_workspace.execute_command.call_count == 1 mock_workspace.execute_command.assert_called_once_with( 'git config --global user.name "Test User"', '/workspace/project' ) @pytest.mark.asyncio async def test_configure_git_user_settings_only_email(mock_workspace): """Test configuring only git user email.""" user_info = MockUserInfo(git_user_name=None, git_user_email='test@example.com') service, _ = _create_service_with_mock_user_context(user_info) await service._configure_git_user_settings(mock_workspace) # Verify only user.email was configured assert mock_workspace.execute_command.call_count == 1 mock_workspace.execute_command.assert_called_once_with( 'git config --global user.email "test@example.com"', '/workspace/project' ) @pytest.mark.asyncio async def test_configure_git_user_settings_neither_set(mock_workspace): """Test when neither git user name nor email is set.""" user_info = MockUserInfo(git_user_name=None, git_user_email=None) service, _ = _create_service_with_mock_user_context(user_info) await service._configure_git_user_settings(mock_workspace) # Verify no git config commands were executed mock_workspace.execute_command.assert_not_called() @pytest.mark.asyncio async def test_configure_git_user_settings_empty_strings(mock_workspace): """Test when git user name and email are empty strings.""" user_info = MockUserInfo(git_user_name='', git_user_email='') service, _ = _create_service_with_mock_user_context(user_info) await service._configure_git_user_settings(mock_workspace) # Empty strings are falsy, so no commands should be executed mock_workspace.execute_command.assert_not_called() @pytest.mark.asyncio async def test_configure_git_user_settings_get_user_info_fails(mock_workspace): """Test handling of exception when get_user_info fails.""" user_info = MockUserInfo() service, mock_user_context = _create_service_with_mock_user_context(user_info) mock_user_context.get_user_info = AsyncMock( side_effect=Exception('User info error') ) # Should not raise exception, just log warning await service._configure_git_user_settings(mock_workspace) # Verify no git config commands were executed mock_workspace.execute_command.assert_not_called() @pytest.mark.asyncio async def test_configure_git_user_settings_name_command_fails(mock_workspace): """Test handling when git config user.name command fails.""" user_info = MockUserInfo( git_user_name='Test User', git_user_email='test@example.com' ) service, _ = _create_service_with_mock_user_context(user_info) # Make the first command fail (user.name), second succeed (user.email) mock_workspace.execute_command = AsyncMock( side_effect=[ MockCommandResult(exit_code=1, stderr='Permission denied'), MockCommandResult(exit_code=0), ] ) # Should not raise exception await service._configure_git_user_settings(mock_workspace) # Verify both commands were still attempted assert mock_workspace.execute_command.call_count == 2 @pytest.mark.asyncio async def test_configure_git_user_settings_email_command_fails(mock_workspace): """Test handling when git config user.email command fails.""" user_info = MockUserInfo( git_user_name='Test User', git_user_email='test@example.com' ) service, _ = _create_service_with_mock_user_context(user_info) # Make the first command succeed (user.name), second fail (user.email) mock_workspace.execute_command = AsyncMock( side_effect=[ MockCommandResult(exit_code=0), MockCommandResult(exit_code=1, stderr='Permission denied'), ] ) # Should not raise exception await service._configure_git_user_settings(mock_workspace) # Verify both commands were still attempted assert mock_workspace.execute_command.call_count == 2 @pytest.mark.asyncio async def test_configure_git_user_settings_special_characters_in_name(mock_workspace): """Test git user name with special characters.""" user_info = MockUserInfo( git_user_name="Test O'Brien", git_user_email='test@example.com' ) service, _ = _create_service_with_mock_user_context(user_info) await service._configure_git_user_settings(mock_workspace) # Verify the name is passed with special characters mock_workspace.execute_command.assert_any_call( 'git config --global user.name "Test O\'Brien"', '/workspace/project' ) # ============================================================================= # Tests for load_and_merge_all_skills (updated to use agent-server) # ============================================================================= class TestMergeSkills: """Test _merge_skills method.""" def test_merges_skills_with_no_duplicates(self): """Test merging skill lists with no duplicate names.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object(AppConversationServiceBase, '__abstractmethods__', set()): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context ) skill1 = Mock(spec=Skill) skill1.name = 'skill1' skill2 = Mock(spec=Skill) skill2.name = 'skill2' skill3 = Mock(spec=Skill) skill3.name = 'skill3' skill_lists = [[skill1], [skill2], [skill3]] # Act result = service._merge_skills(skill_lists) # Assert assert len(result) == 3 names = {s.name for s in result} assert names == {'skill1', 'skill2', 'skill3'} def test_merges_skills_with_duplicates_later_wins(self): """Test that later skill lists override earlier ones for duplicate names.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object(AppConversationServiceBase, '__abstractmethods__', set()): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context ) skill1_v1 = Mock(spec=Skill) skill1_v1.name = 'skill1' skill1_v1.version = 'v1' skill1_v2 = Mock(spec=Skill) skill1_v2.name = 'skill1' skill1_v2.version = 'v2' skill2 = Mock(spec=Skill) skill2.name = 'skill2' skill_lists = [[skill1_v1], [skill1_v2, skill2]] # Act result = service._merge_skills(skill_lists) # Assert assert len(result) == 2 skill1_result = next(s for s in result if s.name == 'skill1') assert skill1_result.version == 'v2' class TestLoadAndMergeAllSkills: """Test load_and_merge_all_skills method (updated to use agent-server).""" @pytest.mark.asyncio @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.load_skills_from_agent_server' ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.build_org_config' ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.build_sandbox_config' ) async def test_loads_skills_successfully( self, mock_build_sandbox_config, mock_build_org_config, mock_load_skills, ): """Test successfully loading skills from agent-server.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object(AppConversationServiceBase, '__abstractmethods__', set()): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context ) mock_workspace = AsyncMock() mock_workspace.working_dir = '/workspace' from openhands.app_server.sandbox.sandbox_models import ExposedUrl sandbox = Mock(spec=SandboxInfo) exposed_url = ExposedUrl( name='AGENT_SERVER', url='http://localhost:8000', port=8000 ) sandbox.exposed_urls = [exposed_url] sandbox.session_api_key = 'test-api-key' skill1 = Mock(spec=Skill) skill1.name = 'skill1' skill2 = Mock(spec=Skill) skill2.name = 'skill2' mock_load_skills.return_value = [skill1, skill2] mock_build_org_config.return_value = {'repository': 'owner/repo'} mock_build_sandbox_config.return_value = {'exposed_urls': []} # Act result = await service.load_and_merge_all_skills( sandbox, 'owner/repo', '/workspace/repo', 'http://localhost:8000' ) # Assert assert len(result) == 2 assert result[0].name == 'skill1' assert result[1].name == 'skill2' mock_load_skills.assert_called_once() call_kwargs = mock_load_skills.call_args[1] assert call_kwargs['agent_server_url'] == 'http://localhost:8000' assert call_kwargs['session_api_key'] == 'test-api-key' assert call_kwargs['project_dir'] == '/workspace/repo' @pytest.mark.asyncio @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.load_skills_from_agent_server' ) async def test_returns_empty_list_when_no_agent_server_url(self, mock_load_skills): """Test returns empty list when agent-server URL is not available.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object(AppConversationServiceBase, '__abstractmethods__', set()): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context ) AsyncMock() from openhands.app_server.sandbox.sandbox_models import ExposedUrl sandbox = Mock(spec=SandboxInfo) exposed_url = ExposedUrl( name='VSCODE', url='http://localhost:8080', port=8080 ) sandbox.exposed_urls = [exposed_url] # Act - pass empty string to simulate no agent server URL # This should still call load_skills_from_agent_server but it will fail result = await service.load_and_merge_all_skills( sandbox, 'owner/repo', '/workspace/repo', '' ) # Assert - should return empty list when agent_server_url is empty assert result == [] @pytest.mark.asyncio @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.load_skills_from_agent_server' ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.build_org_config' ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.build_sandbox_config' ) async def test_uses_project_dir_when_no_repository( self, mock_build_sandbox_config, mock_build_org_config, mock_load_skills, ): """Test uses project_dir directly when no repository is selected.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object(AppConversationServiceBase, '__abstractmethods__', set()): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context ) AsyncMock() from openhands.app_server.sandbox.sandbox_models import ExposedUrl sandbox = Mock(spec=SandboxInfo) exposed_url = ExposedUrl( name='AGENT_SERVER', url='http://localhost:8000', port=8000 ) sandbox.exposed_urls = [exposed_url] sandbox.session_api_key = 'test-key' mock_load_skills.return_value = [] mock_build_org_config.return_value = None mock_build_sandbox_config.return_value = None # Act await service.load_and_merge_all_skills( sandbox, None, '/workspace', 'http://localhost:8000' ) # Assert call_kwargs = mock_load_skills.call_args[1] assert call_kwargs['project_dir'] == '/workspace' @pytest.mark.asyncio @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.load_skills_from_agent_server' ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.build_org_config' ) @patch( 'openhands.app_server.app_conversation.app_conversation_service_base.build_sandbox_config' ) async def test_handles_exception_gracefully( self, mock_build_sandbox_config, mock_build_org_config, mock_load_skills, ): """Test handles exceptions during skill loading.""" # Arrange mock_user_context = Mock(spec=UserContext) with patch.object(AppConversationServiceBase, '__abstractmethods__', set()): service = AppConversationServiceBase( init_git_in_empty_workspace=True, user_context=mock_user_context ) AsyncMock() from openhands.app_server.sandbox.sandbox_models import ExposedUrl sandbox = Mock(spec=SandboxInfo) exposed_url = ExposedUrl( name='AGENT_SERVER', url='http://localhost:8000', port=8000 ) sandbox.exposed_urls = [exposed_url] sandbox.session_api_key = 'test-key' mock_load_skills.side_effect = Exception('Network error') # Act result = await service.load_and_merge_all_skills( sandbox, 'owner/repo', '/workspace/repo', 'http://localhost:8000' ) # Assert assert result == []