| """ |
| Telegram Bot集成测试 |
| """ |
| |
| import pytest |
| import tempfile |
| from unittest.mock import Mock, AsyncMock, patch |
| |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) |
| |
| from claude_agent.telegram.bot import TelegramBot |
| |
| |
| class TestTelegramBotIntegration: |
| """Telegram Bot集成测试""" |
| |
| @pytest.fixture |
| def temp_config_dir(self): |
| """创建临时配置目录""" |
| with tempfile.TemporaryDirectory() as tmpdir: |
| yield tmpdir |
| |
| @pytest.fixture |
| def mock_config(self, temp_config_dir): |
| """创建测试配置""" |
| config_content = """ |
| [telegram] |
| bot_token = "123456:TEST_TOKEN" |
| allowed_users = [12345] |
| allowed_groups = [-67890] |
| |
| [telegram.message] |
| stream_update_interval = 0.1 |
| max_message_length = 4096 |
| enable_markdown = true |
| context_history_limit = 10 |
| |
| [telegram.files] |
| supported_image_formats = ["jpg", "png"] |
| supported_document_formats = ["txt", "pdf"] |
| max_file_size_mb = 5 |
| temp_dir = "temp/test" |
| |
| [agent] |
| default_mode = "interactive" |
| max_conversation_history = 100 |
| |
| [logging] |
| level = "INFO" |
| colored = true |
| """ |
| config_file = Path(temp_config_dir) / "test.toml" |
| config_file.write_text(config_content) |
| return temp_config_dir |
| |
| @pytest.mark.asyncio |
| async def test_bot_initialization(self, mock_config): |
| """测试Bot初始化""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| # Mock配置管理器 |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456:TEST_TOKEN', |
| 'allowed_users': [12345], |
| 'allowed_groups': [-67890], |
| 'message': { |
| 'stream_update_interval': 0.1, |
| 'context_history_limit': 10 |
| }, |
| 'files': { |
| 'temp_dir': '/tmp/test', |
| 'max_file_size_mb': 5 |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot |
| bot = TelegramBot(config_name="test") |
| |
| # 验证配置加载 |
| assert bot.telegram_config['bot_token'] == '123456:TEST_TOKEN' |
| assert 12345 in bot.telegram_config['allowed_users'] |
| assert -67890 in bot.telegram_config['allowed_groups'] |
| |
| @pytest.mark.asyncio |
| async def test_bot_invalid_token(self, mock_config): |
| """测试无效Token的处理""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': 'YOUR_BOT_TOKEN_HERE', # 无效token |
| 'allowed_users': [12345], |
| 'allowed_groups': [] |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 应该抛出配置错误 |
| with pytest.raises(ValueError, match="请配置有效的Telegram Bot Token"): |
| TelegramBot(config_name="test") |
| |
| @pytest.mark.asyncio |
| async def test_bot_missing_config(self): |
| """测试缺少配置的处理""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456:TEST_TOKEN', |
| # 缺少 allowed_users 和 allowed_groups |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 应该抛出配置错误 |
| with pytest.raises(ValueError, match="缺少必需的配置项"): |
| TelegramBot(config_name="test") |
| |
| @pytest.mark.asyncio |
| async def test_component_integration(self, mock_config): |
| """测试组件集成""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| # Mock配置 |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456:TEST_TOKEN', |
| 'allowed_users': [12345], |
| 'allowed_groups': [-67890], |
| 'message': { |
| 'stream_update_interval': 0.1, |
| 'context_history_limit': 10 |
| }, |
| 'files': { |
| 'temp_dir': '/tmp/test', |
| 'max_file_size_mb': 5 |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # Mock Application创建 |
| with patch('claude_agent.telegram.bot.Application') as mock_app_class: |
| mock_builder = Mock() |
| mock_app = Mock() |
| mock_app.bot = Mock() |
| mock_app.initialize = AsyncMock() |
| mock_app.start = AsyncMock() |
| mock_app.updater = Mock() |
| mock_app.updater.start_polling = AsyncMock() |
| mock_app.add_handler = Mock() |
| mock_app.add_error_handler = Mock() |
| |
| mock_builder.token.return_value = mock_builder |
| mock_builder.build.return_value = mock_app |
| mock_app_class.builder.return_value = mock_builder |
| |
| # 创建并初始化Bot |
| bot = TelegramBot(config_name="test") |
| await bot.initialize() |
| |
| # 验证组件创建 |
| assert bot.context_manager is not None |
| assert bot.file_handler is not None |
| assert bot.claude_agent is not None |
| assert bot.stream_sender is not None |
| assert bot.message_handler is not None |
| assert bot.telegram_client is not None |
| |
| # 验证Application配置 |
| mock_builder.token.assert_called_with('123456:TEST_TOKEN') |
| mock_app.add_handler.assert_called() |
| mock_app.add_error_handler.assert_called() |
| |
| @pytest.mark.asyncio |
| async def test_dependency_injection(self, mock_config): |
| """测试依赖注入功能""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456:TEST_TOKEN', |
| 'allowed_users': [12345], |
| 'allowed_groups': [-67890], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp', 'max_file_size_mb': 5} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot |
| bot = TelegramBot(config_name="test") |
| |
| # 创建Mock组件 |
| mock_context_manager = Mock() |
| mock_file_handler = Mock() |
| mock_claude_agent = Mock() |
| mock_stream_sender = Mock() |
| |
| # 注入依赖 |
| bot.set_context_manager(mock_context_manager) |
| bot.set_file_handler(mock_file_handler) |
| bot.set_claude_agent(mock_claude_agent) |
| bot.set_stream_sender(mock_stream_sender) |
| |
| # 验证依赖注入 |
| assert bot.context_manager == mock_context_manager |
| assert bot.file_handler == mock_file_handler |
| assert bot.claude_agent == mock_claude_agent |
| assert bot.stream_sender == mock_stream_sender |
| |
| @pytest.mark.asyncio |
| async def test_get_stats(self, mock_config): |
| """测试获取统计信息""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456:TEST_TOKEN', |
| 'allowed_users': [111, 222], |
| 'allowed_groups': [-333, -444], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp', 'max_file_size_mb': 5} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot |
| bot = TelegramBot(config_name="test") |
| |
| # Mock上下文管理器 |
| mock_context_manager = Mock() |
| mock_context_manager.get_chat_count.return_value = 5 |
| bot.set_context_manager(mock_context_manager) |
| |
| # 获取统计信息 |
| stats = bot.get_stats() |
| |
| # 验证统计信息 |
| assert stats['is_running'] is False |
| assert stats['config_name'] == "test" |
| assert stats['allowed_users_count'] == 2 |
| assert stats['allowed_groups_count'] == 2 |
| assert stats['active_chats'] == 5 |
| |
| @pytest.mark.asyncio |
| async def test_send_admin_message(self, mock_config): |
| """测试发送管理员消息""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456:TEST_TOKEN', |
| 'allowed_users': [111, 222], |
| 'allowed_groups': [], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp', 'max_file_size_mb': 5} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot |
| bot = TelegramBot(config_name="test") |
| |
| # Mock Telegram客户端 |
| mock_client = Mock() |
| mock_client.send_message = AsyncMock() |
| bot.telegram_client = mock_client |
| |
| # 发送管理员消息 |
| await bot.send_admin_message("Test admin message") |
| |
| # 验证发送给所有允许的用户 |
| assert mock_client.send_message.call_count == 2 |
| calls = mock_client.send_message.call_args_list |
| |
| # 验证消息内容 |
| for call in calls: |
| assert call[1]['text'].startswith("📢 管理员消息:") |
| assert "Test admin message" in call[1]['text'] |
| |
| # 验证发送给正确的用户 |
| sent_to_users = {call[1]['chat_id'] for call in calls} |
| assert sent_to_users == {111, 222} |
| |
| @pytest.mark.asyncio |
| async def test_configuration_validation_edge_cases(self): |
| """测试配置验证的边缘情况""" |
| with patch('claude_agent.utils.config.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| |
| # 测试空Token |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '', |
| 'allowed_users': [123], |
| 'allowed_groups': [] |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| with pytest.raises(ValueError, match="请配置有效的Telegram Bot Token"): |
| TelegramBot(config_name="test") |
| |
| # 测试空的allowed_users和allowed_groups |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123:valid_token', |
| 'allowed_users': [], |
| 'allowed_groups': [] |
| } |
| |
| # 应该能够创建（虽然没有授权用户） |
| bot = TelegramBot(config_name="test") |
| assert bot.telegram_config['allowed_users'] == [] |
| assert bot.telegram_config['allowed_groups'] == [] |