| """ |
| CLI增强界面键盘绑定和高级功能测试 |
| 专注于提升enhanced_interface.py的覆盖率 |
| """ |
| |
| import pytest |
| import asyncio |
| from unittest.mock import Mock, AsyncMock, patch, MagicMock |
| |
| from src.claude_agent.cli.enhanced_interface import EnhancedCLIInterface |
| from src.claude_agent.core.agent import ThinkingMode |
| |
| # 设置所有异步测试为自动标记 |
| pytestmark = pytest.mark.asyncio |
| |
| |
| class TestEnhancedCLIKeyBindings: |
| """测试增强CLI键盘绑定""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = EnhancedCLIInterface() |
| |
| def test_key_bindings_setup(self): |
| """测试键盘绑定设置""" |
| # 验证键盘绑定对象存在 |
| assert self.cli.key_bindings is not None |
| |
| # 验证绑定了预期的快捷键 |
| bindings = self.cli.key_bindings.bindings |
| |
| # 基本验证键盘绑定被设置 |
| assert len(bindings) > 0 |
| |
| @patch('src.claude_agent.cli.enhanced_interface.Progress') |
| @patch('src.claude_agent.cli.enhanced_interface.AgentCore') |
| @patch('src.claude_agent.cli.enhanced_interface.MCPToolIntegration') |
| @patch('src.claude_agent.cli.enhanced_interface.create_sshout_integration') |
| async def test_initialize_with_sshout(self, mock_sshout_factory, mock_mcp, mock_agent_core, mock_progress): |
| """测试带SSHOUT的完整初始化""" |
| # 设置Mock |
| mock_agent = Mock() |
| mock_agent_core.return_value = mock_agent |
| |
| mock_mcp_instance = AsyncMock() |
| mock_mcp.return_value = mock_mcp_instance |
| |
| mock_sshout_instance = AsyncMock() |
| mock_sshout_factory.return_value = mock_sshout_instance |
| |
| # 设置Progress Mock |
| progress_instance = Mock() |
| progress_context = Mock() |
| progress_context.__enter__ = Mock(return_value=progress_instance) |
| progress_context.__exit__ = Mock(return_value=None) |
| mock_progress.return_value = progress_context |
| progress_instance.add_task = Mock(return_value="task_id") |
| progress_instance.update = Mock() |
| |
| # 执行初始化 |
| await self.cli.initialize(api_key="test-key") |
| |
| # 验证所有组件都被初始化 |
| mock_agent_core.assert_called_once_with(api_key="test-key") |
| mock_mcp.assert_called_once_with(mock_agent) |
| mock_sshout_factory.assert_called_once_with(mock_agent) |
| |
| assert self.cli.agent == mock_agent |
| assert self.cli.mcp_integration == mock_mcp_instance |
| assert self.cli.sshout_integration == mock_sshout_instance |
| |
| @patch('src.claude_agent.cli.enhanced_interface.Progress') |
| @patch('src.claude_agent.cli.enhanced_interface.AgentCore') |
| async def test_initialize_agent_failure(self, mock_agent_core, mock_progress): |
| """测试Agent初始化失败""" |
| # 设置Agent抛出异常 |
| mock_agent_core.side_effect = Exception("API Key invalid") |
| |
| # 设置Progress Mock |
| progress_instance = Mock() |
| progress_context = Mock() |
| progress_context.__enter__ = Mock(return_value=progress_instance) |
| progress_context.__exit__ = Mock(return_value=None) |
| mock_progress.return_value = progress_context |
| progress_instance.add_task = Mock(return_value="task_id") |
| progress_instance.update = Mock() |
| |
| # 执行初始化 - 应该捕获异常，打印错误信息，然后重新抛出 |
| with patch.object(self.cli.console, 'print') as mock_print: |
| with pytest.raises(Exception, match="API Key invalid"): |
| await self.cli.initialize(api_key="invalid-key") |
| # 验证打印了错误信息 |
| mock_print.assert_called() |
| |
| # 验证失败状态 |
| assert self.cli.agent is None |
| assert self.cli.mcp_integration is None |
| assert self.cli.sshout_integration is None |
| |
| |
| class TestEnhancedCLIUserInput: |
| """测试增强CLI用户输入处理""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = EnhancedCLIInterface() |
| |
| async def test_get_user_input_success(self): |
| """测试成功获取用户输入""" |
| # 设置Mock会话 |
| mock_session = AsyncMock() |
| mock_session.prompt_async.return_value = "test input" |
| self.cli.session = mock_session |
| |
| result = await self.cli.get_user_input("Enter text") |
| |
| assert result == "test input" |
| mock_session.prompt_async.assert_called_once() |
| |
| async def test_get_user_input_keyboard_interrupt(self): |
| """测试键盘中断处理""" |
| # 设置Mock会话抛出KeyboardInterrupt |
| mock_session = AsyncMock() |
| mock_session.prompt_async.side_effect = KeyboardInterrupt() |
| self.cli.session = mock_session |
| |
| result = await self.cli.get_user_input("Enter text") |
| |
| assert result == "" |
| |
| async def test_get_user_input_eof_error(self): |
| """测试EOF错误处理""" |
| # 设置Mock会话抛出EOFError |
| mock_session = AsyncMock() |
| mock_session.prompt_async.side_effect = EOFError() |
| self.cli.session = mock_session |
| |
| result = await self.cli.get_user_input("Enter text") |
| |
| assert result == "" |
| |
| async def test_confirm_action_yes(self): |
| """测试确认操作返回是""" |
| mock_session = AsyncMock() |
| mock_session.prompt_async.return_value = "y" |
| self.cli.session = mock_session |
| |
| result = await self.cli.confirm_action("Are you sure?") |
| |
| assert result is True |
| |
| async def test_confirm_action_no(self): |
| """测试确认操作返回否""" |
| mock_session = AsyncMock() |
| mock_session.prompt_async.return_value = "n" |
| self.cli.session = mock_session |
| |
| result = await self.cli.confirm_action("Are you sure?") |
| |
| assert result is False |
| |
| async def test_confirm_action_keyboard_interrupt(self): |
| """测试确认操作键盘中断""" |
| mock_session = AsyncMock() |
| mock_session.prompt_async.side_effect = KeyboardInterrupt() |
| self.cli.session = mock_session |
| |
| result = await self.cli.confirm_action("Are you sure?") |
| |
| assert result is False |
| |
| |
| class TestEnhancedCLIAdvancedFeatures: |
| """测试增强CLI高级功能""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = EnhancedCLIInterface() |
| |
| async def test_shutdown_with_components(self): |
| """测试有组件时的关闭""" |
| # 设置Mock组件 |
| mock_mcp = AsyncMock() |
| self.cli.mcp_integration = mock_mcp |
| |
| # Mock _disconnect_sshout 方法 |
| self.cli._disconnect_sshout = AsyncMock() |
| mock_sshout = Mock() |
| self.cli.sshout_integration = mock_sshout |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.shutdown() |
| |
| # 验证关闭调用 |
| mock_mcp.shutdown.assert_called_once() |
| self.cli._disconnect_sshout.assert_called_once_with(show_message=False) |
| mock_print.assert_called() |
| |
| async def test_shutdown_without_components(self): |
| """测试无组件时的关闭""" |
| self.cli.mcp_integration = None |
| self.cli.sshout_integration = None |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.shutdown() |
| mock_print.assert_called() |
| |
| async def test_shutdown_with_exception(self): |
| """测试关闭时的异常处理""" |
| # 设置Mock组件抛出异常 |
| mock_mcp = AsyncMock() |
| mock_mcp.shutdown.side_effect = Exception("Shutdown failed") |
| self.cli.mcp_integration = mock_mcp |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| # 关闭时应该传播异常，除非有异常处理 |
| with pytest.raises(Exception, match="Shutdown failed"): |
| await self.cli.shutdown() |
| |
| def test_show_command_history_empty(self): |
| """测试空命令历史显示""" |
| # 设置空历史 |
| self.cli.history._storage = [] |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_command_history() |
| mock_print.assert_called() |
| # 检查是否显示了空历史信息 |
| args = mock_print.call_args[0] |
| assert "[yellow]" in str(args[0]) |
| |
| def test_show_command_history_with_data(self): |
| """测试有数据的命令历史显示""" |
| # 设置有历史数据 |
| self.cli.history._storage = ["command1", "command2", "command3"] |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_command_history() |
| # 验证调用了多次print（标题+历史命令） |
| assert mock_print.call_count >= 1 |
| |
| def test_show_conversation_history_no_agent(self): |
| """测试无Agent时显示对话历史""" |
| self.cli.agent = None |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_conversation_history() |
| mock_print.assert_called_once() |
| args = mock_print.call_args[0] |
| assert "[red]" in str(args[0]) |
| |
| def test_show_conversation_history_empty(self): |
| """测试空对话历史显示""" |
| mock_agent = Mock() |
| mock_agent.get_conversation_history.return_value = [] |
| self.cli.agent = mock_agent |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_conversation_history() |
| mock_print.assert_called_once() |
| args = mock_print.call_args[0] |
| assert "[yellow]" in str(args[0]) |
| |
| def test_show_conversation_history_with_data(self): |
| """测试有对话历史时的显示""" |
| mock_agent = Mock() |
| mock_agent.get_conversation_history.return_value = [ |
| {"role": "user", "content": "Hello"}, |
| {"role": "assistant", "content": "Hi there! How can I help you today?"} |
| ] |
| self.cli.agent = mock_agent |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_conversation_history() |
| # 验证调用了多次print（标题+历史消息） |
| assert mock_print.call_count >= 2 |
| |
| async def test_switch_mode_no_agent(self): |
| """测试无Agent时的模式切换""" |
| self.cli.agent = None |
| |
| # 这个方法直接访问self.agent.thinking_mode，没有None检查 |
| # 应该引发AttributeError |
| with pytest.raises(AttributeError): |
| await self.cli.switch_mode() |
| |
| async def test_switch_mode_with_confirm(self): |
| """测试模式切换（确认）""" |
| mock_agent = Mock() |
| mock_agent.thinking_mode = ThinkingMode.INTERACTIVE |
| self.cli.agent = mock_agent |
| |
| # Mock confirm_action 返回True |
| self.cli.confirm_action = AsyncMock(return_value=True) |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.switch_mode() |
| mock_agent.set_thinking_mode.assert_called_once_with(ThinkingMode.YOLO) |
| mock_print.assert_called() |
| |
| async def test_switch_mode_without_confirm(self): |
| """测试模式切换（取消）""" |
| mock_agent = Mock() |
| mock_agent.thinking_mode = ThinkingMode.INTERACTIVE |
| self.cli.agent = mock_agent |
| |
| # Mock confirm_action 返回False |
| self.cli.confirm_action = AsyncMock(return_value=False) |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.switch_mode() |
| mock_agent.set_thinking_mode.assert_not_called() |
| mock_print.assert_called() |
| |
| |
| class TestEnhancedCLICommands: |
| """测试命令处理功能""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = EnhancedCLIInterface() |
| |
| async def test_process_command_quit(self): |
| """测试退出命令""" |
| result = await self.cli.process_command("/quit") |
| assert result is False |
| |
| async def test_process_command_exit(self): |
| """测试退出命令""" |
| result = await self.cli.process_command("/exit") |
| assert result is False |
| |
| async def test_process_command_help(self): |
| """测试帮助命令""" |
| with patch.object(self.cli, 'show_help') as mock_help: |
| result = await self.cli.process_command("/help") |
| mock_help.assert_called_once() |
| assert result is True |
| |
| async def test_process_command_status(self): |
| """测试状态命令""" |
| with patch.object(self.cli, 'show_status') as mock_status: |
| result = await self.cli.process_command("/status") |
| mock_status.assert_called_once() |
| assert result is True |
| |
| async def test_process_command_tools(self): |
| """测试工具命令""" |
| with patch.object(self.cli, 'show_tools') as mock_tools: |
| result = await self.cli.process_command("/tools") |
| mock_tools.assert_called_once() |
| assert result is True |
| |
| async def test_process_command_history(self): |
| """测试历史命令""" |
| with patch.object(self.cli, 'show_command_history') as mock_cmd_history: |
| with patch.object(self.cli, 'show_conversation_history') as mock_conv_history: |
| result = await self.cli.process_command("/history") |
| mock_cmd_history.assert_called_once() |
| mock_conv_history.assert_called_once() |
| assert result is True |
| |
| async def test_process_command_non_command(self): |
| """测试非命令输入""" |
| with patch.object(self.cli, 'process_user_input') as mock_process: |
| result = await self.cli.process_command("What is Python?") |
| mock_process.assert_called_once_with("What is Python?") |
| assert result is True |
| |
| |
| if __name__ == '__main__': |
| pytest.main([__file__]) |