| """ |
| Telegram Bot核心业务逻辑单元测试 |
| 专注测试配置验证、组件初始化、生命周期管理等核心功能 |
| """ |
| |
| import pytest |
| import asyncio |
| import signal |
| from unittest.mock import Mock, AsyncMock, patch, MagicMock |
| from telegram import Bot as TelegramBotAPI, User |
| from telegram.ext import Application |
| |
| 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 TestTelegramBotCore: |
| """Telegram Bot核心功能测试""" |
| |
| @pytest.fixture |
| def valid_config_manager(self): |
| """创建有效的配置管理器""" |
| config_manager = Mock() |
| telegram_config = { |
| 'bot_token': '123456789:ABCDEFGHijklmnopqrstuvwxyz-123456789', |
| 'allowed_users': [111111, 222222], |
| 'allowed_groups': [-333333], |
| 'message': { |
| 'context_history_limit': 50, |
| 'stream_update_interval': 1.0 |
| }, |
| 'files': { |
| 'temp_dir': 'temp/telegram', |
| 'max_file_size_mb': 20 |
| } |
| } |
| webhook_config = { |
| 'enabled': False |
| } |
| config_manager.get_telegram_config.return_value = telegram_config |
| config_manager.get_webhook_config.return_value = webhook_config |
| return config_manager |
| |
| @pytest.fixture |
| def invalid_config_manager(self): |
| """创建无效的配置管理器""" |
| config_manager = Mock() |
| # 缺少必需的配置项 |
| telegram_config = { |
| 'bot_token': '123456789:ABCDEFGHijklmnopqrstuvwxyz-123456789' |
| # 缺少 allowed_users 和 allowed_groups |
| } |
| config_manager.get_telegram_config.return_value = telegram_config |
| return config_manager |
| |
| @pytest.fixture |
| def empty_token_config_manager(self): |
| """创建空token的配置管理器""" |
| config_manager = Mock() |
| telegram_config = { |
| 'bot_token': 'YOUR_BOT_TOKEN_HERE', # 默认无效token |
| 'allowed_users': [111111], |
| 'allowed_groups': [-333333] |
| } |
| config_manager.get_telegram_config.return_value = telegram_config |
| return config_manager |
| |
| def test_initialization_with_valid_config(self, valid_config_manager): |
| """测试使用有效配置初始化""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot(config_name="test") |
| |
| assert bot.config_name == "test" |
| assert bot.config_manager == valid_config_manager |
| assert bot.telegram_config == valid_config_manager.get_telegram_config() |
| assert not bot.is_running |
| assert bot.application is None |
| assert bot.webhook_integration is None |
| |
| def test_initialization_with_invalid_config(self, invalid_config_manager): |
| """测试使用无效配置初始化应该抛出异常""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=invalid_config_manager): |
| with pytest.raises(ValueError, match="缺少必需的配置项"): |
| TelegramBot(config_name="invalid") |
| |
| def test_initialization_with_empty_token(self, empty_token_config_manager): |
| """测试使用空token配置初始化应该抛出异常""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=empty_token_config_manager): |
| with pytest.raises(ValueError, match="请配置有效的Telegram Bot Token"): |
| TelegramBot(config_name="empty_token") |
| |
| def test_config_validation_missing_required_keys(self): |
| """测试配置验证：缺少必需的键""" |
| config_manager = Mock() |
| |
| # 测试缺少 bot_token |
| telegram_config = { |
| 'allowed_users': [111111], |
| 'allowed_groups': [-333333] |
| } |
| config_manager.get_telegram_config.return_value = telegram_config |
| |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=config_manager): |
| with pytest.raises(ValueError, match="缺少必需的配置项: telegram.bot_token"): |
| TelegramBot() |
| |
| def test_config_validation_all_missing_keys(self): |
| """测试配置验证：所有必需的键都缺少""" |
| config_manager = Mock() |
| telegram_config = {} # 空配置 |
| config_manager.get_telegram_config.return_value = telegram_config |
| |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=config_manager): |
| with pytest.raises(ValueError) as exc_info: |
| TelegramBot() |
| |
| # 应该抛出第一个缺少的配置项错误 |
| assert "缺少必需的配置项" in str(exc_info.value) |
| |
| @pytest.mark.asyncio |
| async def test_initialize_components_success(self, valid_config_manager): |
| """测试组件初始化成功""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| # Mock Application.builder() |
| mock_application = Mock(spec=Application) |
| mock_bot_api = Mock(spec=TelegramBotAPI) |
| mock_user = Mock(spec=User) |
| mock_user.id = 12345 |
| mock_user.username = "test_bot" |
| |
| mock_bot_api.get_me = AsyncMock(return_value=mock_user) |
| mock_application.bot = mock_bot_api |
| |
| mock_builder = Mock() |
| mock_builder.token.return_value = mock_builder |
| mock_builder.build.return_value = mock_application |
| |
| with patch('claude_agent.telegram.bot.Application.builder', return_value=mock_builder): |
| # Mock所有组件创建 |
| with patch('claude_agent.telegram.bot.TelegramClientAdapter') as mock_client_adapter: |
| with patch('claude_agent.telegram.bot.ContextManager') as mock_context_manager: |
| with patch('claude_agent.telegram.bot.FileHandler') as mock_file_handler: |
| with patch('claude_agent.telegram.bot.ClaudeAgentAdapter') as mock_claude_adapter: |
| with patch('claude_agent.telegram.bot.StreamMessageSender') as mock_stream_sender: |
| with patch('claude_agent.telegram.bot.MessageHandler') as mock_message_handler: |
| # Mock webhook初始化 |
| with patch.object(bot, '_initialize_webhook_integration', new_callable=AsyncMock): |
| with patch.object(bot, '_process_webhook_message_queue', new_callable=AsyncMock): |
| with patch.object(bot, '_register_handlers'): |
| |
| await bot.initialize() |
| |
| # 验证组件创建 |
| assert bot.application == mock_application |
| mock_client_adapter.assert_called_once_with(mock_bot_api) |
| mock_context_manager.assert_called_once() |
| mock_file_handler.assert_called_once() |
| mock_claude_adapter.assert_called_once() |
| mock_stream_sender.assert_called_once() |
| mock_message_handler.assert_called_once() |
| |
| @pytest.mark.asyncio |
| async def test_initialize_components_bot_api_failure(self, valid_config_manager): |
| """测试Bot API获取失败的情况""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_application = Mock(spec=Application) |
| mock_bot_api = Mock(spec=TelegramBotAPI) |
| mock_bot_api.get_me = AsyncMock(side_effect=Exception("API Error")) |
| mock_application.bot = mock_bot_api |
| |
| mock_builder = Mock() |
| mock_builder.token.return_value = mock_builder |
| mock_builder.build.return_value = mock_application |
| |
| with patch('claude_agent.telegram.bot.Application.builder', return_value=mock_builder): |
| with pytest.raises(Exception, match="API Error"): |
| await bot.initialize() |
| |
| @pytest.mark.asyncio |
| async def test_start_already_running(self, valid_config_manager): |
| """测试启动已经在运行的Bot""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| bot.is_running = True # 设置为已运行状态 |
| |
| # Mock telegram_client 避免stop方法中发送通知失败 |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock() |
| bot.telegram_client = mock_telegram_client |
| |
| with patch('asyncio.get_running_loop'): |
| await bot.start() |
| |
| # 由于start方法的finally块会调用stop，所以最终is_running会变为False |
| # 但是我们可以验证没有进行初始化 |
| assert bot.application is None # 未进行初始化 |
| |
| @pytest.mark.asyncio |
| async def test_start_initialization_failure(self, valid_config_manager): |
| """测试启动时初始化失败""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| with patch.object(bot, 'initialize', side_effect=Exception("Init failed")): |
| with patch.object(bot, 'stop', new_callable=AsyncMock) as mock_stop: |
| with pytest.raises(Exception, match="Init failed"): |
| await bot.start() |
| |
| # 应该调用stop方法清理 |
| mock_stop.assert_called_once() |
| |
| @pytest.mark.asyncio |
| async def test_stop_not_running(self, valid_config_manager): |
| """测试停止未运行的Bot""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| bot.is_running = False |
| |
| # 应该能正常调用而不出错 |
| await bot.stop() |
| assert not bot.is_running |
| |
| @pytest.mark.asyncio |
| async def test_stop_with_cleanup(self, valid_config_manager): |
| """测试停止Bot并清理资源""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| bot.is_running = True |
| |
| # Mock应用程序 |
| mock_application = Mock() |
| mock_updater = Mock() |
| mock_updater.stop = AsyncMock() |
| mock_application.updater = mock_updater |
| mock_application.stop = AsyncMock() |
| mock_application.shutdown = AsyncMock() |
| bot.application = mock_application |
| |
| # Mock组件 - 使用AsyncMock |
| mock_file_handler = Mock() |
| mock_file_handler.cleanup_temp_files = AsyncMock() # 修复：使用AsyncMock |
| bot.file_handler = mock_file_handler |
| |
| mock_context_manager = Mock() |
| mock_context_manager.cleanup_old_chats = Mock() |
| bot.context_manager = mock_context_manager |
| |
| # Mock webhook集成 |
| mock_webhook_integration = Mock() |
| mock_webhook_integration.stop = AsyncMock() |
| bot.webhook_integration = mock_webhook_integration |
| |
| # Mock webhook队列任务 - 创建一个真正的已取消任务 |
| async def dummy_task(): |
| await asyncio.sleep(10) # 长时间睡眠的任务 |
| |
| webhook_task = asyncio.create_task(dummy_task()) |
| webhook_task.cancel() |
| bot.webhook_queue_task = webhook_task |
| |
| with patch.object(bot, '_send_shutdown_notification', new_callable=AsyncMock): |
| await bot.stop() |
| |
| # 验证清理调用 |
| mock_updater.stop.assert_called_once() |
| mock_application.stop.assert_called_once() |
| mock_application.shutdown.assert_called_once() |
| mock_webhook_integration.stop.assert_called_once() |
| mock_file_handler.cleanup_temp_files.assert_called_once() |
| mock_context_manager.cleanup_old_chats.assert_called_once() |
| # 任务已被取消，不需要验证cancel调用 |
| assert not bot.is_running |
| |
| def test_signal_handlers_setup(self, valid_config_manager): |
| """测试信号处理器设置""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| with patch('signal.signal') as mock_signal: |
| bot._setup_signal_handlers() |
| |
| # 验证SIGINT和SIGTERM信号都被设置 |
| expected_calls = [ |
| ((signal.SIGINT, mock_signal.call_args_list[0][0][1]),), |
| ((signal.SIGTERM, mock_signal.call_args_list[1][0][1]),) |
| ] |
| assert mock_signal.call_count == 2 |
| |
| def test_signal_handler_functionality(self, valid_config_manager): |
| """测试信号处理器功能""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| # 验证shutdown_event初始状态 |
| assert not bot.shutdown_event.is_set() |
| |
| with patch('signal.signal') as mock_signal: |
| bot._setup_signal_handlers() |
| |
| # 获取信号处理函数 |
| signal_handler = mock_signal.call_args_list[0][0][1] |
| |
| # 模拟收到信号 |
| signal_handler(signal.SIGINT, None) |
| |
| # 验证shutdown_event被设置 |
| assert bot.shutdown_event.is_set() |
| |
| @pytest.mark.asyncio |
| async def test_send_startup_notification_success(self, valid_config_manager): |
| """测试启动通知发送成功""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock() |
| bot.telegram_client = mock_telegram_client |
| |
| await bot._send_startup_notification() |
| |
| # 验证发送给所有允许的用户 |
| expected_calls = len(bot.telegram_config['allowed_users']) |
| assert mock_telegram_client.send_message.call_count == expected_calls |
| |
| # 验证消息内容 |
| call_args = mock_telegram_client.send_message.call_args_list[0] |
| assert "Claude Telegram Bot 已启动" in call_args[1]['text'] |
| |
| @pytest.mark.asyncio |
| async def test_send_startup_notification_partial_failure(self, valid_config_manager): |
| """测试启动通知部分失败""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_telegram_client = Mock() |
| # 第一个用户发送成功，第二个用户发送失败 |
| mock_telegram_client.send_message = AsyncMock(side_effect=[None, Exception("Send failed")]) |
| bot.telegram_client = mock_telegram_client |
| |
| # 应该不抛出异常，只记录警告 |
| await bot._send_startup_notification() |
| |
| assert mock_telegram_client.send_message.call_count == 2 |
| |
| @pytest.mark.asyncio |
| async def test_send_shutdown_notification_success(self, valid_config_manager): |
| """测试关闭通知发送成功""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock() |
| bot.telegram_client = mock_telegram_client |
| |
| await bot._send_shutdown_notification() |
| |
| # 验证发送给所有允许的用户 |
| expected_calls = len(bot.telegram_config['allowed_users']) |
| assert mock_telegram_client.send_message.call_count == expected_calls |
| |
| # 验证消息内容 |
| call_args = mock_telegram_client.send_message.call_args_list[0] |
| assert "Claude Telegram Bot 正在关闭" in call_args[1]['text'] |
| |
| def test_get_stats_basic(self, valid_config_manager): |
| """测试获取基本统计信息""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| stats = bot.get_stats() |
| |
| assert stats['is_running'] is False |
| assert stats['config_name'] == "local" # 默认值 |
| assert stats['allowed_users_count'] == 2 |
| assert stats['allowed_groups_count'] == 1 |
| |
| def test_get_stats_with_context_manager(self, valid_config_manager): |
| """测试获取包含上下文管理器的统计信息""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_context_manager = Mock() |
| mock_context_manager.get_chat_count.return_value = 5 |
| bot.context_manager = mock_context_manager |
| |
| stats = bot.get_stats() |
| |
| assert stats['active_chats'] == 5 |
| |
| @pytest.mark.asyncio |
| async def test_send_admin_message_success(self, valid_config_manager): |
| """测试发送管理员消息成功""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock() |
| bot.telegram_client = mock_telegram_client |
| |
| test_message = "System maintenance at 3AM" |
| await bot.send_admin_message(test_message) |
| |
| # 验证发送给所有允许的用户 |
| expected_calls = len(bot.telegram_config['allowed_users']) |
| assert mock_telegram_client.send_message.call_count == expected_calls |
| |
| # 验证消息格式 |
| call_args = mock_telegram_client.send_message.call_args_list[0] |
| assert "📢 管理员消息:" in call_args[1]['text'] |
| assert test_message in call_args[1]['text'] |
| |
| @pytest.mark.asyncio |
| async def test_send_admin_message_failure(self, valid_config_manager): |
| """测试发送管理员消息失败""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock(side_effect=Exception("Send failed")) |
| bot.telegram_client = mock_telegram_client |
| |
| # 应该不抛出异常，只记录错误 |
| await bot.send_admin_message("Test message") |
| |
| # 应该尝试发送给所有用户 |
| expected_calls = len(bot.telegram_config['allowed_users']) |
| assert mock_telegram_client.send_message.call_count == expected_calls |
| |
| def test_dependency_injection_methods(self, valid_config_manager): |
| """测试依赖注入方法（用于测试）""" |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=valid_config_manager): |
| bot = TelegramBot() |
| |
| 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 |
| |
| |
| class TestTelegramBotErrorHandling: |
| """Telegram Bot错误处理测试""" |
| |
| @pytest.fixture |
| def bot_with_valid_config(self): |
| """创建有效配置的Bot实例""" |
| config_manager = Mock() |
| telegram_config = { |
| 'bot_token': '123456789:ValidToken', |
| 'allowed_users': [111], |
| 'allowed_groups': [-222] |
| } |
| config_manager.get_telegram_config.return_value = telegram_config |
| config_manager.get_webhook_config.return_value = {'enabled': False} |
| |
| with patch('claude_agent.telegram.bot.get_config_manager', return_value=config_manager): |
| return TelegramBot() |
| |
| @pytest.mark.asyncio |
| async def test_error_handler_with_chat(self, bot_with_valid_config): |
| """测试错误处理器（有聊天信息）""" |
| bot = bot_with_valid_config |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock() |
| bot.telegram_client = mock_telegram_client |
| |
| # 创建Mock Update |
| mock_update = Mock() |
| mock_chat = Mock() |
| mock_chat.id = 123456 |
| mock_update.effective_chat = mock_chat |
| |
| # 创建Mock Context |
| mock_context = Mock() |
| mock_context.error = Exception("Test error") |
| |
| await bot._error_handler(mock_update, mock_context) |
| |
| # 验证发送错误消息 |
| mock_telegram_client.send_message.assert_called_once() |
| call_args = mock_telegram_client.send_message.call_args |
| assert call_args[1]['chat_id'] == 123456 |
| assert "处理消息时出现内部错误" in call_args[1]['text'] |
| |
| @pytest.mark.asyncio |
| async def test_error_handler_without_chat(self, bot_with_valid_config): |
| """测试错误处理器（无聊天信息）""" |
| bot = bot_with_valid_config |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock() |
| bot.telegram_client = mock_telegram_client |
| |
| # 无Update或无effective_chat |
| mock_context = Mock() |
| mock_context.error = Exception("Test error") |
| |
| await bot._error_handler(None, mock_context) |
| |
| # 不应该发送消息 |
| mock_telegram_client.send_message.assert_not_called() |
| |
| @pytest.mark.asyncio |
| async def test_error_handler_send_message_fails(self, bot_with_valid_config): |
| """测试错误处理器发送消息失败""" |
| bot = bot_with_valid_config |
| |
| mock_telegram_client = Mock() |
| mock_telegram_client.send_message = AsyncMock(side_effect=Exception("Send failed")) |
| bot.telegram_client = mock_telegram_client |
| |
| # 创建Mock Update |
| mock_update = Mock() |
| mock_chat = Mock() |
| mock_chat.id = 123456 |
| mock_update.effective_chat = mock_chat |
| |
| mock_context = Mock() |
| mock_context.error = Exception("Original error") |
| |
| # 应该不抛出异常，只记录错误 |
| await bot._error_handler(mock_update, mock_context) |
| |
| mock_telegram_client.send_message.assert_called_once() |