| """ |
| Telegram Bot黑盒测试 |
| 端到端功能验证，不依赖内部实现细节 |
| """ |
| |
| import pytest |
| import tempfile |
| import os |
| import sys |
| from pathlib import Path |
| from unittest.mock import Mock, patch, AsyncMock |
| import asyncio |
| import json |
| |
| # 添加src目录到Python路径 |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src")) |
| |
| from claude_agent.telegram.bot import TelegramBot |
| |
| |
| class TestTelegramBotBlackBox: |
| """Telegram Bot黑盒测试""" |
| |
| @pytest.fixture |
| def temp_config_file(self): |
| """创建临时配置文件""" |
| config_content = """ |
| [telegram] |
| bot_token = "123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789" |
| allowed_users = [12345, 67890] |
| allowed_groups = [-100123, -200456] |
| |
| [telegram.message] |
| stream_update_interval = 0.1 |
| max_message_length = 4096 |
| enable_markdown = true |
| context_history_limit = 20 |
| |
| [telegram.files] |
| supported_image_formats = ["jpg", "jpeg", "png", "gif"] |
| supported_document_formats = ["pdf", "txt", "md"] |
| max_file_size_mb = 10 |
| temp_dir = "temp/blackbox_test" |
| |
| [agent] |
| default_mode = "interactive" |
| max_conversation_history = 50 |
| |
| [logging] |
| level = "INFO" |
| colored = true |
| """ |
| with tempfile.NamedTemporaryFile(mode='w', suffix='.toml', delete=False) as f: |
| f.write(config_content) |
| return f.name |
| |
| def test_bot_creation_with_valid_config(self, temp_config_file): |
| """测试使用有效配置创建Bot""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| # Mock配置管理器 |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345, 67890], |
| 'allowed_groups': [-100123, -200456], |
| 'message': { |
| 'stream_update_interval': 0.1, |
| 'context_history_limit': 20 |
| }, |
| 'files': { |
| 'temp_dir': '/tmp/blackbox_test', |
| 'max_file_size_mb': 10 |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot应该成功 |
| bot = TelegramBot("test") |
| |
| # 验证基本属性 |
| assert bot.config_name == "test" |
| assert not bot.is_running |
| |
| # 验证统计信息 |
| stats = bot.get_stats() |
| assert stats['config_name'] == "test" |
| assert stats['allowed_users_count'] == 2 |
| assert stats['allowed_groups_count'] == 2 |
| assert stats['is_running'] is False |
| |
| def test_bot_creation_with_invalid_token(self): |
| """测试使用无效Token创建Bot""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| # 使用空字符串，这会触发验证错误 |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '', # 空字符串会失败 |
| 'allowed_users': [12345], |
| 'allowed_groups': [] |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 应该抛出配置错误 |
| with pytest.raises(ValueError, match="请配置有效的Telegram Bot Token"): |
| TelegramBot("test") |
| |
| def test_bot_creation_missing_required_config(self): |
| """测试缺少必需配置项""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| # 缺少 allowed_users 和 allowed_groups |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 应该抛出配置错误 |
| with pytest.raises(ValueError, match="缺少必需的配置项"): |
| TelegramBot("test") |
| |
| @pytest.mark.asyncio |
| async def test_bot_initialization_components(self): |
| """测试Bot组件初始化""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345], |
| 'allowed_groups': [-67890], |
| 'message': {'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.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("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('123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789') |
| |
| def test_bot_stats_reporting(self): |
| """测试Bot统计信息报告""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [111, 222, 333], |
| 'allowed_groups': [-444, -555], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp', 'max_file_size_mb': 5} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| bot = TelegramBot("production") |
| |
| # Mock上下文管理器 |
| mock_context_manager = Mock() |
| mock_context_manager.get_chat_count.return_value = 15 |
| bot.set_context_manager(mock_context_manager) |
| |
| stats = bot.get_stats() |
| |
| # 验证统计信息 |
| expected_stats = { |
| 'is_running': False, |
| 'config_name': 'production', |
| 'allowed_users_count': 3, |
| 'allowed_groups_count': 2, |
| 'active_chats': 15 |
| } |
| |
| for key, value in expected_stats.items(): |
| assert stats[key] == value |
| |
| @pytest.mark.asyncio |
| async def test_admin_message_broadcasting(self): |
| """测试管理员消息广播""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [111, 222, 333], |
| 'allowed_groups': [-444], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp', 'max_file_size_mb': 5} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| bot = TelegramBot("test") |
| |
| # Mock Telegram客户端 |
| mock_client = Mock() |
| mock_client.send_message = AsyncMock() |
| bot.telegram_client = mock_client |
| |
| # 发送管理员消息 |
| test_message = "System maintenance at 2AM" |
| await bot.send_admin_message(test_message) |
| |
| # 验证向所有允许用户发送消息 |
| assert mock_client.send_message.call_count == 3 |
| calls = mock_client.send_message.call_args_list |
| |
| # 验证发送给正确的用户 |
| sent_to_users = {call[1]['chat_id'] for call in calls} |
| assert sent_to_users == {111, 222, 333} |
| |
| # 验证消息格式 |
| for call in calls: |
| message_text = call[1]['text'] |
| assert "📢 管理员消息:" in message_text |
| assert test_message in message_text |
| |
| def test_dependency_injection_functionality(self): |
| """测试依赖注入功能""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| '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 = TelegramBot("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 |
| |
| def test_configuration_validation_edge_cases(self): |
| """测试配置验证边缘情况""" |
| with patch('claude_agent.telegram.bot.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): |
| TelegramBot("test") |
| |
| # 测试默认Token |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': 'YOUR_BOT_TOKEN_HERE', |
| 'allowed_users': [123], |
| 'allowed_groups': [] |
| } |
| |
| with pytest.raises(ValueError): |
| TelegramBot("test") |
| |
| # 测试空的用户和群组列表（应该允许） |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [], |
| 'allowed_groups': [] |
| } |
| |
| # 应该成功创建，但没有授权用户 |
| bot = TelegramBot("test") |
| assert bot.telegram_config['allowed_users'] == [] |
| assert bot.telegram_config['allowed_groups'] == [] |
| |
| @pytest.mark.asyncio |
| async def test_bot_error_handling(self): |
| """测试Bot错误处理""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345], |
| '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 = TelegramBot("test") |
| |
| # 测试发送管理员消息时的错误处理 |
| mock_client = Mock() |
| mock_client.send_message = AsyncMock(side_effect=Exception("Network error")) |
| bot.telegram_client = mock_client |
| |
| # 应该不抛出异常，只记录日志 |
| await bot.send_admin_message("Test message") |
| |
| def test_config_driven_behavior(self): |
| """测试配置驱动的行为""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| |
| # 测试不同的配置值 |
| configs = [ |
| { |
| 'bot_token': '111:AAA', |
| 'allowed_users': [1, 2, 3, 4, 5], |
| 'allowed_groups': [-1, -2], |
| 'message': {'context_history_limit': 100}, |
| 'files': {'temp_dir': '/tmp/large', 'max_file_size_mb': 50} |
| }, |
| { |
| 'bot_token': '222:BBB', |
| 'allowed_users': [100], |
| 'allowed_groups': [-100, -200, -300], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp/small', 'max_file_size_mb': 1} |
| } |
| ] |
| |
| for i, config in enumerate(configs): |
| mock_config_manager.get_telegram_config.return_value = config |
| mock_get_config.return_value = mock_config_manager |
| |
| bot = TelegramBot(f"test_{i}") |
| stats = bot.get_stats() |
| |
| # 验证配置正确反映在统计中 |
| assert stats['allowed_users_count'] == len(config['allowed_users']) |
| assert stats['allowed_groups_count'] == len(config['allowed_groups']) |
| assert bot.telegram_config['bot_token'] == config['bot_token'] |
| |
| def test_bot_lifecycle_state_management(self): |
| """测试Bot生命周期状态管理""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345], |
| '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 = TelegramBot("test") |
| |
| # 初始状态 |
| assert not bot.is_running |
| stats = bot.get_stats() |
| assert stats['is_running'] is False |
| |
| # 模拟状态变化（在实际启动中会改变） |
| bot.is_running = True |
| stats = bot.get_stats() |
| assert stats['is_running'] is True |
| |
| def test_multiple_bot_instances_independence(self): |
| """测试多个Bot实例的独立性""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| |
| # 创建两个不同配置的Bot |
| config1 = { |
| 'bot_token': '111:AAA', |
| 'allowed_users': [111], |
| 'allowed_groups': [-111], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp/bot1', 'max_file_size_mb': 5} |
| } |
| |
| config2 = { |
| 'bot_token': '222:BBB', |
| 'allowed_users': [222, 333], |
| 'allowed_groups': [-222], |
| 'message': {'context_history_limit': 20}, |
| 'files': {'temp_dir': '/tmp/bot2', 'max_file_size_mb': 10} |
| } |
| |
| # 创建第一个Bot |
| mock_config_manager.get_telegram_config.return_value = config1 |
| mock_get_config.return_value = mock_config_manager |
| bot1 = TelegramBot("bot1") |
| |
| # 创建第二个Bot |
| mock_config_manager.get_telegram_config.return_value = config2 |
| bot2 = TelegramBot("bot2") |
| |
| # 验证两个Bot配置独立 |
| stats1 = bot1.get_stats() |
| stats2 = bot2.get_stats() |
| |
| assert stats1['config_name'] == "bot1" |
| assert stats2['config_name'] == "bot2" |
| assert stats1['allowed_users_count'] == 1 |
| assert stats2['allowed_users_count'] == 2 |
| assert bot1.telegram_config['bot_token'] == '111:AAA' |
| assert bot2.telegram_config['bot_token'] == '222:BBB' |
| |
| def test_file_path_resolution(self): |
| """测试文件路径处理""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345], |
| 'allowed_groups': [], |
| 'message': {'context_history_limit': 10}, |
| 'files': { |
| 'temp_dir': 'relative/path/telegram', # 相对路径 |
| 'max_file_size_mb': 5 |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| bot = TelegramBot("test") |
| |
| # 验证配置被保存（当前实现没有路径解析） |
| temp_dir = bot.telegram_config['files']['temp_dir'] |
| assert temp_dir == 'relative/path/telegram' # 保持原始值 |
| |
| |
| class TestTelegramBotIntegrationBlackBox: |
| """Telegram Bot集成黑盒测试""" |
| |
| def test_end_to_end_configuration_loading(self): |
| """端到端配置加载测试""" |
| # 创建临时配置文件 |
| config_content = """ |
| [telegram] |
| bot_token = "987654321:ZYXWVUTSRQPONMLKJIHGFEDCBAxyz987654" |
| allowed_users = [111, 222] |
| allowed_groups = [-333] |
| |
| [telegram.message] |
| stream_update_interval = 2.0 |
| max_message_length = 2048 |
| context_history_limit = 25 |
| |
| [telegram.files] |
| max_file_size_mb = 15 |
| temp_dir = "test_temp" |
| |
| [agent] |
| default_mode = "yolo" |
| |
| [logging] |
| level = "DEBUG" |
| """ |
| |
| with tempfile.NamedTemporaryFile(mode='w', suffix='.toml', delete=False) as f: |
| f.write(config_content) |
| temp_file = f.name |
| |
| try: |
| # Mock配置管理器直接加载文件 |
| if sys.version_info >= (3, 11): |
| import tomllib |
| else: |
| import tomli as tomllib |
| |
| with open(temp_file, 'rb') as f: |
| loaded_config = tomllib.load(f) |
| |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = loaded_config['telegram'] |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot并验证配置加载 |
| bot = TelegramBot("integration_test") |
| |
| # 验证配置正确加载 |
| assert bot.telegram_config['bot_token'] == "987654321:ZYXWVUTSRQPONMLKJIHGFEDCBAxyz987654" |
| assert bot.telegram_config['allowed_users'] == [111, 222] |
| assert bot.telegram_config['allowed_groups'] == [-333] |
| assert bot.telegram_config['message']['stream_update_interval'] == 2.0 |
| assert bot.telegram_config['message']['max_message_length'] == 2048 |
| assert bot.telegram_config['files']['max_file_size_mb'] == 15 |
| |
| finally: |
| os.unlink(temp_file) |
| |
| def test_complete_component_interaction(self): |
| """完整组件交互测试""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345], |
| 'allowed_groups': [-67890], |
| 'message': {'context_history_limit': 10}, |
| 'files': {'temp_dir': '/tmp/integration', 'max_file_size_mb': 5} |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| # 创建Bot |
| bot = TelegramBot("integration") |
| |
| # 创建Mock组件并注入 |
| mock_context_manager = Mock() |
| mock_file_handler = Mock() |
| mock_claude_agent = Mock() |
| mock_stream_sender = Mock() |
| |
| # 配置Mock行为 |
| mock_context_manager.get_chat_count.return_value = 5 |
| mock_context_manager.get_context.return_value = [] |
| |
| 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) |
| |
| # 验证组件协同工作 |
| stats = bot.get_stats() |
| assert stats['active_chats'] == 5 |
| |
| # 验证依赖关系正确建立 |
| assert bot.context_manager == mock_context_manager |
| assert bot.file_handler == mock_file_handler |
| |
| def test_production_readiness_checklist(self): |
| """生产就绪性检查清单""" |
| with patch('claude_agent.telegram.bot.get_config_manager') as mock_get_config: |
| mock_config_manager = Mock() |
| mock_config_manager.get_telegram_config.return_value = { |
| 'bot_token': '123456789:ABCDEFGHIJKLMNOPQRSTUVWXYZ123456789', |
| 'allowed_users': [12345], |
| 'allowed_groups': [-67890], |
| 'message': { |
| 'stream_update_interval': 1.0, |
| 'max_message_length': 4096, |
| 'context_history_limit': 50 |
| }, |
| 'files': { |
| 'temp_dir': '/var/tmp/telegram_bot', |
| 'max_file_size_mb': 20, |
| 'supported_image_formats': ['jpg', 'png', 'gif'], |
| 'supported_document_formats': ['pdf', 'txt', 'md'] |
| } |
| } |
| mock_get_config.return_value = mock_config_manager |
| |
| bot = TelegramBot("production") |
| |
| # 生产就绪性检查 |
| checks = { |
| 'valid_token_format': bot.telegram_config['bot_token'].count(':') == 1, |
| 'has_authorized_users': len(bot.telegram_config['allowed_users']) > 0, |
| 'reasonable_file_size_limit': bot.telegram_config['files']['max_file_size_mb'] <= 50, |
| 'reasonable_context_limit': bot.telegram_config['message']['context_history_limit'] <= 100, |
| 'valid_update_interval': 0.1 <= bot.telegram_config['message']['stream_update_interval'] <= 5.0 |
| } |
| |
| # 所有检查都应该通过 |
| for check_name, result in checks.items(): |
| assert result, f"Production readiness check failed: {check_name}" |
| |
| # 验证统计信息完整 |
| stats = bot.get_stats() |
| required_stats = ['is_running', 'config_name', 'allowed_users_count', 'allowed_groups_count'] |
| for stat in required_stats: |
| assert stat in stats, f"Missing required stat: {stat}" |