| """ |
| CLI接口模块全面测试 |
| 测试基础CLI接口的所有功能 |
| """ |
| |
| import pytest |
| import asyncio |
| from unittest.mock import Mock, AsyncMock, patch, MagicMock |
| from io import StringIO |
| import sys |
| |
| from src.claude_agent.cli.interface import CLIInterface |
| from src.claude_agent.core.agent import ThinkingMode |
| |
| |
| class TestCLIInterfaceBasic: |
| """测试CLI接口基础功能""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = CLIInterface() |
| |
| def test_initialization(self): |
| """测试CLI初始化""" |
| assert self.cli.console is not None |
| assert self.cli.agent is None |
| assert self.cli.mcp_integration is None |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Progress') |
| @patch('src.claude_agent.cli.interface.AgentCore') |
| @patch('src.claude_agent.cli.interface.MCPToolIntegration') |
| async def test_initialize_success(self, mock_mcp, mock_agent_core, mock_progress): |
| """测试成功初始化""" |
| # 设置Mock |
| mock_agent = Mock() |
| mock_agent_core.return_value = mock_agent |
| |
| mock_mcp_instance = AsyncMock() |
| mock_mcp.return_value = mock_mcp_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_mcp_instance.setup_default_tools.assert_called_once() |
| mock_mcp_instance.enhance_agent_with_tools.assert_called_once() |
| |
| assert self.cli.agent == mock_agent |
| assert self.cli.mcp_integration == mock_mcp_instance |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Progress') |
| @patch('src.claude_agent.cli.interface.AgentCore') |
| async def test_initialize_failure(self, mock_agent_core, mock_progress): |
| """测试初始化失败""" |
| # 设置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") |
| |
| # 执行初始化 |
| await self.cli.initialize(api_key="invalid-key") |
| |
| # 验证失败状态 |
| assert self.cli.agent is None |
| assert self.cli.mcp_integration is None |
| |
| def test_show_welcome(self): |
| """测试显示欢迎信息""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_welcome() |
| mock_print.assert_called_once() |
| |
| def test_show_help(self): |
| """测试显示帮助信息""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_help() |
| mock_print.assert_called_once() |
| |
| def test_show_status_no_agent(self): |
| """测试无Agent时的状态显示""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_status() |
| mock_print.assert_called_once() |
| |
| def test_show_status_with_agent(self): |
| """测试有Agent时的状态显示""" |
| # 创建Mock Agent |
| mock_agent = Mock() |
| mock_agent.thinking_mode = ThinkingMode.INTERACTIVE |
| mock_agent.conversation_history = ["msg1", "msg2"] # 设置为列表 |
| self.cli.agent = mock_agent |
| |
| # 创建Mock MCP |
| mock_mcp = Mock() |
| mock_tool_manager = Mock() |
| mock_tool_manager.tools = ["tool1", "tool2"] |
| mock_mcp.tool_manager = mock_tool_manager |
| self.cli.mcp_integration = mock_mcp |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_status() |
| mock_print.assert_called_once() |
| |
| def test_show_history_no_agent(self): |
| """测试无Agent时的历史显示""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_history() |
| mock_print.assert_called_once() |
| # 检查是否显示了错误信息 |
| args = mock_print.call_args[0] |
| assert "[red]" in str(args[0]) |
| |
| def test_show_history_with_agent(self): |
| """测试有Agent时的历史显示""" |
| mock_agent = Mock() |
| mock_agent.get_conversation_history.return_value = [ |
| {"role": "user", "content": "Hello"}, |
| {"role": "assistant", "content": "Hi there!"} |
| ] |
| self.cli.agent = mock_agent |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_history() |
| # 验证调用了print（可能多次，取决于消息数量） |
| assert mock_print.call_count >= 1 |
| |
| def test_show_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_history() |
| mock_print.assert_called_once() |
| args = mock_print.call_args[0] |
| assert "[yellow]" in str(args[0]) |
| |
| @pytest.mark.asyncio |
| async def test_switch_mode_no_agent(self): |
| """测试无Agent时的模式切换""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.switch_mode() |
| mock_print.assert_called_once() |
| # 检查是否显示了错误信息 |
| args = mock_print.call_args[0] |
| assert "[red]" in str(args[0]) |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Confirm.ask') |
| async def test_switch_mode_with_agent_confirm(self, mock_confirm): |
| """测试有Agent时的模式切换（确认）""" |
| mock_agent = Mock() |
| mock_agent.thinking_mode = ThinkingMode.INTERACTIVE |
| self.cli.agent = mock_agent |
| |
| mock_confirm.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_once() |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Confirm.ask') |
| async def test_switch_mode_with_agent_cancel(self, mock_confirm): |
| """测试有Agent时的模式切换（取消）""" |
| mock_agent = Mock() |
| mock_agent.thinking_mode = ThinkingMode.INTERACTIVE |
| self.cli.agent = mock_agent |
| |
| mock_confirm.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_once() |
| |
| def test_show_tools_no_mcp(self): |
| """测试无MCP时的工具显示""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_tools() |
| mock_print.assert_called_once() |
| # 检查是否显示了错误信息 |
| args = mock_print.call_args[0] |
| assert "[red]" in str(args[0]) |
| |
| def test_show_tools_with_mcp_empty(self): |
| """测试有MCP但无工具时的显示""" |
| mock_mcp = Mock() |
| mock_tool_manager = Mock() |
| mock_tool_manager.get_available_tools.return_value = [] |
| mock_mcp.tool_manager = mock_tool_manager |
| self.cli.mcp_integration = mock_mcp |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_tools() |
| mock_print.assert_called_once() |
| args = mock_print.call_args[0] |
| assert "[yellow]" in str(args[0]) |
| |
| def test_show_tools_with_mcp_tools(self): |
| """测试有MCP和工具时的显示""" |
| mock_mcp = Mock() |
| mock_tool_manager = Mock() |
| mock_tool_manager.get_available_tools.return_value = ["tool1", "tool2"] |
| mock_tool_manager.get_tool_info.return_value = {"description": "Test tool"} |
| mock_mcp.tool_manager = mock_tool_manager |
| self.cli.mcp_integration = mock_mcp |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| self.cli.show_tools() |
| mock_print.assert_called_once() |
| |
| |
| class TestCLIInterfaceCommands: |
| """测试CLI命令处理""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = CLIInterface() |
| |
| @pytest.mark.asyncio |
| async def test_process_command_help(self): |
| """测试help命令""" |
| 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 |
| |
| @pytest.mark.asyncio |
| async def test_process_command_status(self): |
| """测试status命令""" |
| 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 |
| |
| @pytest.mark.asyncio |
| async def test_process_command_history(self): |
| """测试history命令""" |
| with patch.object(self.cli, 'show_history') as mock_history: |
| result = await self.cli.process_command("history") |
| mock_history.assert_called_once() |
| assert result is True |
| |
| @pytest.mark.asyncio |
| async def test_process_command_tools(self): |
| """测试tools命令""" |
| 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 |
| |
| @pytest.mark.asyncio |
| async def test_process_command_quit(self): |
| """测试quit命令""" |
| result = await self.cli.process_command("quit") |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_process_command_exit(self): |
| """测试exit命令""" |
| result = await self.cli.process_command("exit") |
| assert result is False |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Confirm.ask') |
| async def test_process_command_clear_no_agent(self, mock_confirm): |
| """测试无Agent时的clear命令""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| result = await self.cli.process_command("clear") |
| mock_print.assert_called_once() |
| args = mock_print.call_args[0] |
| assert "[red]" in str(args[0]) |
| assert result is True |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Confirm.ask') |
| async def test_process_command_clear_with_agent_confirm(self, mock_confirm): |
| """测试有Agent时的clear命令（确认）""" |
| mock_agent = Mock() |
| self.cli.agent = mock_agent |
| mock_confirm.return_value = True |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| result = await self.cli.process_command("clear") |
| mock_agent.clear_history.assert_called_once() |
| mock_print.assert_called_once() |
| assert result is True |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Confirm.ask') |
| async def test_process_command_clear_with_agent_cancel(self, mock_confirm): |
| """测试有Agent时的clear命令（取消）""" |
| mock_agent = Mock() |
| self.cli.agent = mock_agent |
| mock_confirm.return_value = False |
| |
| result = await self.cli.process_command("clear") |
| mock_agent.clear_history.assert_not_called() |
| assert result is True |
| |
| @pytest.mark.asyncio |
| async def test_process_command_mode(self): |
| """测试mode命令""" |
| with patch.object(self.cli, 'switch_mode') as mock_mode: |
| result = await self.cli.process_command("mode") |
| mock_mode.assert_called_once() |
| assert result is True |
| |
| @pytest.mark.asyncio |
| async def test_process_command_unknown(self): |
| """测试未知命令""" |
| with patch.object(self.cli, 'process_user_input') as mock_process: |
| result = await self.cli.process_command("unknown_command") |
| mock_process.assert_called_once_with("unknown_command") |
| assert result is True |
| |
| |
| class TestCLIInterfaceUserInput: |
| """测试用户输入处理""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = CLIInterface() |
| |
| @pytest.mark.asyncio |
| async def test_process_user_input_no_agent(self): |
| """测试无Agent时的查询处理""" |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.process_user_input("What is Python?") |
| mock_print.assert_called_once() |
| # 检查是否显示了错误信息 |
| args = mock_print.call_args[0] |
| assert "[red]" in str(args[0]) |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Progress') |
| async def test_process_user_input_with_agent_interactive(self, mock_progress): |
| """测试有Agent时的交互模式查询处理""" |
| mock_agent = AsyncMock() |
| mock_agent.thinking_mode = ThinkingMode.INTERACTIVE |
| mock_agent.process_user_input.return_value = "Python is a programming language" |
| self.cli.agent = mock_agent |
| |
| # 设置Mock MCP |
| mock_mcp = AsyncMock() |
| mock_mcp.process_tool_calls_in_response.return_value = "Enhanced response" |
| self.cli.mcp_integration = mock_mcp |
| |
| # 设置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") |
| |
| with patch.object(self.cli.console, 'print') as mock_print: |
| await self.cli.process_user_input("What is Python?") |
| mock_agent.process_user_input.assert_called_once_with("What is Python?") |
| mock_mcp.process_tool_calls_in_response.assert_called_once() |
| mock_print.assert_called() |
| |
| @pytest.mark.asyncio |
| @patch('src.claude_agent.cli.interface.Progress') |
| async def test_process_user_input_with_agent_yolo(self, mock_progress): |
| """测试有Agent时的YOLO模式查询处理""" |
| mock_agent = AsyncMock() |
| mock_agent.thinking_mode = ThinkingMode.YOLO |
| mock_agent.process_user_input.return_value = "Python is a programming language" |
| self.cli.agent = mock_agent |
| |
| # 设置Mock MCP |
| mock_mcp = AsyncMock() |
| mock_mcp.process_tool_calls_in_response.return_value = "Enhanced response" |
| self.cli.mcp_integration = mock_mcp |
| |
| # 设置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 patch('logging.getLogger') as mock_logger: |
| await self.cli.process_user_input("What is Python?") |
| mock_agent.process_user_input.assert_called_once_with("What is Python?") |
| |
| |
| class TestCLIInterfaceIntegration: |
| """测试CLI集成功能""" |
| |
| def setup_method(self): |
| """测试前准备""" |
| self.cli = CLIInterface() |
| |
| def test_setup_logging(self): |
| """测试日志设置""" |
| # 验证日志设置不会抛出异常 |
| self.cli.setup_logging() |
| # 基本验证：console对象存在 |
| assert self.cli.console is not None |