| """ |
| 消息处理器核心功能单元测试 |
| 重点测试群聊参与管理、权限验证、消息处理等核心业务逻辑 |
| """ |
| |
| import pytest |
| import random |
| from unittest.mock import Mock, AsyncMock, patch |
| from telegram import Update, Message, User, Chat |
| |
| import sys |
| from pathlib import Path |
| sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src")) |
| |
| from claude_agent.telegram.message_handler import MessageHandler, GroupParticipationManager |
| |
| |
| class TestGroupParticipationManager: |
| """群聊参与管理器测试""" |
| |
| def test_initialization_default(self): |
| """测试默认初始化""" |
| manager = GroupParticipationManager() |
| |
| # 检查默认bot名字 |
| expected_names = [ |
| 'claude', 'Claude', 'CLAUDE', |
| '克劳德', 'AI', 'ai', 'bot', 'Bot', 'BOT', |
| '助手', '机器人' |
| ] |
| assert manager._bot_names == expected_names |
| assert manager._participation_range == [1, 10] |
| assert manager._message_counters == {} |
| assert manager._trigger_thresholds == {} |
| |
| def test_initialization_with_config(self): |
| """测试使用配置初始化""" |
| bot_names = ['TestBot', 'MyBot'] |
| participation_range = [5, 15] |
| |
| manager = GroupParticipationManager( |
| bot_names=bot_names, |
| participation_range=participation_range |
| ) |
| |
| assert manager._bot_names == bot_names |
| assert manager._participation_range == participation_range |
| |
| def test_reset_counter(self): |
| """测试重置计数器""" |
| manager = GroupParticipationManager(participation_range=[3, 5]) |
| chat_id = "-123456789" |
| |
| with patch('random.randint', return_value=4) as mock_randint: |
| manager.reset_counter(chat_id) |
| |
| assert manager._message_counters[chat_id] == 0 |
| assert manager._trigger_thresholds[chat_id] == 4 |
| mock_randint.assert_called_once_with(3, 5) |
| |
| def test_should_participate_random_first_time(self): |
| """测试首次检查随机参与""" |
| manager = GroupParticipationManager() |
| chat_id = "-123456789" |
| |
| with patch.object(manager, 'reset_counter') as mock_reset: |
| result = manager.should_participate_random(chat_id) |
| |
| assert result is False |
| mock_reset.assert_called_once_with(chat_id) |
| |
| def test_should_participate_random_threshold_not_reached(self): |
| """测试未达到参与阈值""" |
| manager = GroupParticipationManager() |
| chat_id = "-123456789" |
| |
| # 手动设置状态 |
| manager._message_counters[chat_id] = 2 |
| manager._trigger_thresholds[chat_id] = 5 |
| |
| result = manager.should_participate_random(chat_id) |
| |
| assert result is False |
| assert manager._message_counters[chat_id] == 3 # 计数器递增 |
| |
| def test_should_participate_random_threshold_reached(self): |
| """测试达到参与阈值""" |
| manager = GroupParticipationManager() |
| chat_id = "-123456789" |
| |
| # 手动设置状态 |
| manager._message_counters[chat_id] = 4 |
| manager._trigger_thresholds[chat_id] = 5 |
| |
| with patch.object(manager, 'reset_counter') as mock_reset: |
| result = manager.should_participate_random(chat_id) |
| |
| assert result is True |
| mock_reset.assert_called_once_with(chat_id) |
| |
| def test_is_name_mentioned_positive(self): |
| """测试名字被提及的情况""" |
| manager = GroupParticipationManager() |
| |
| test_cases = [ |
| "Hello Claude, how are you?", |
| "CLAUDE please help", |
| "I need 助手 to assist", |
| "Call the AI for help", |
| "克劳德 你好" |
| ] |
| |
| for text in test_cases: |
| result = manager.is_name_mentioned(text) |
| assert result is True, f"Failed for text: {text}" |
| |
| def test_is_name_mentioned_negative(self): |
| """测试名字未被提及的情况""" |
| manager = GroupParticipationManager() |
| |
| test_cases = [ |
| "Hello world", |
| "This is a normal message", |
| "Just chatting with friends", |
| "No machines here" # 不包含任何默认bot名字的词 |
| ] |
| |
| for text in test_cases: |
| result = manager.is_name_mentioned(text) |
| assert result is False, f"Failed for text: {text}" |
| |
| def test_is_name_mentioned_case_insensitive(self): |
| """测试大小写不敏感的名字检测""" |
| manager = GroupParticipationManager(bot_names=['TestBot']) |
| |
| test_cases = [ |
| "testbot help me", |
| "TESTBOT please", |
| "TestBot is here", |
| "I need testBot" |
| ] |
| |
| for text in test_cases: |
| result = manager.is_name_mentioned(text) |
| assert result is True, f"Failed for text: {text}" |
| |
| |
| class TestMessageHandler: |
| """消息处理器测试""" |
| |
| @pytest.fixture |
| def mock_components(self): |
| """创建所有组件的Mock""" |
| return { |
| 'telegram_client': Mock(), |
| 'context_manager': Mock(), |
| 'file_handler': Mock(), |
| 'claude_agent': Mock(), |
| 'stream_sender': Mock() |
| } |
| |
| @pytest.fixture |
| def message_handler(self, mock_components): |
| """创建消息处理器实例""" |
| return MessageHandler( |
| telegram_client=mock_components['telegram_client'], |
| context_manager=mock_components['context_manager'], |
| file_handler=mock_components['file_handler'], |
| claude_agent=mock_components['claude_agent'], |
| stream_sender=mock_components['stream_sender'], |
| allowed_users=[111111, 222222], |
| allowed_groups=[-333333, -444444] |
| ) |
| |
| def test_initialization_default_config(self, mock_components): |
| """测试默认配置初始化""" |
| handler = MessageHandler( |
| allowed_users=[111], |
| allowed_groups=[-222], |
| **mock_components |
| ) |
| |
| assert handler.allowed_users == {111} |
| assert handler.allowed_groups == {-222} |
| assert handler.webhook_broadcast_callback is None |
| assert isinstance(handler.participation_manager, GroupParticipationManager) |
| |
| def test_initialization_with_config(self, mock_components): |
| """测试使用配置初始化""" |
| telegram_config = { |
| 'group_participation': { |
| 'bot_names': ['MyBot', 'TestBot'], |
| 'random_participation_range': [8, 15] |
| } |
| } |
| |
| webhook_callback = Mock() |
| |
| handler = MessageHandler( |
| allowed_users=[111], |
| allowed_groups=[-222], |
| telegram_config=telegram_config, |
| webhook_broadcast_callback=webhook_callback, |
| **mock_components |
| ) |
| |
| assert handler.webhook_broadcast_callback == webhook_callback |
| assert handler.participation_manager._bot_names == ['MyBot', 'TestBot'] |
| assert handler.participation_manager._participation_range == [8, 15] |
| |
| def test_build_user_info(self, message_handler): |
| """测试构建用户信息""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 123456 |
| mock_user.username = "test_user" |
| mock_user.first_name = "Test" |
| mock_user.last_name = "User" |
| mock_user.language_code = "en" |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.id = -789012 |
| mock_chat.title = "Test Group" |
| mock_chat.type = "group" |
| |
| user_info = message_handler._build_user_info(mock_user, mock_chat) |
| |
| expected_user_info = { |
| 'user_id': 123456, |
| 'username': "test_user", |
| 'first_name': "Test", |
| 'last_name': "User", |
| 'language_code': "en", # 根据实际实现添加 |
| 'chat_id': -789012, |
| 'chat_title': "Test Group", |
| 'chat_type': "group" |
| } |
| |
| assert user_info == expected_user_info |
| |
| def test_build_user_info_missing_optional_fields(self, message_handler): |
| """测试构建用户信息（缺少可选字段）""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 123456 |
| mock_user.username = None # 可选字段 |
| mock_user.first_name = "Test" |
| mock_user.last_name = None # 可选字段 |
| mock_user.language_code = None |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.id = 111111 # 私聊 |
| mock_chat.title = None # 私聊没有标题 |
| mock_chat.type = "private" |
| |
| user_info = message_handler._build_user_info(mock_user, mock_chat) |
| |
| expected_user_info = { |
| 'user_id': 123456, |
| 'username': None, |
| 'first_name': "Test", |
| 'last_name': None, |
| 'language_code': None, |
| 'chat_id': 111111, |
| 'chat_title': None, |
| 'chat_type': "private" |
| } |
| |
| assert user_info == expected_user_info |
| |
| @pytest.mark.asyncio |
| async def test_save_message_to_agent(self, message_handler, mock_components): |
| """测试保存消息到Agent""" |
| # Mock Claude Agent的内部方法 |
| mock_agent = Mock() |
| mock_components['claude_agent']._get_or_create_agent = Mock(return_value=mock_agent) |
| |
| user_info = {'user_id': 123, 'username': 'test'} |
| |
| await message_handler._save_message_to_agent( |
| chat_id=-123456, |
| user_id=123, |
| message="Test message", |
| is_bot=False, |
| user_info=user_info |
| ) |
| |
| # 验证获取Agent被调用 |
| mock_components['claude_agent']._get_or_create_agent.assert_called_once_with(-123456) |
| |
| @pytest.mark.asyncio |
| async def test_is_authorized_user_private_chat_authorized(self, message_handler): |
| """测试私聊授权用户""" |
| mock_user = Mock() |
| mock_user.id = 111111 # 在允许列表中 |
| |
| mock_chat = Mock() |
| mock_chat.type = "private" |
| mock_chat.id = 111111 |
| |
| mock_update = Mock() |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| result = await message_handler._is_authorized(mock_update) |
| assert result is True |
| |
| @pytest.mark.asyncio |
| async def test_is_authorized_user_private_chat_unauthorized(self, message_handler): |
| """测试私聊未授权用户""" |
| mock_user = Mock() |
| mock_user.id = 999999 # 不在允许列表中 |
| |
| mock_chat = Mock() |
| mock_chat.type = "private" |
| mock_chat.id = 999999 |
| |
| mock_update = Mock() |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| result = await message_handler._is_authorized(mock_update) |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_is_authorized_group_chat_authorized(self, message_handler): |
| """测试授权群组""" |
| mock_user = Mock() |
| mock_user.id = 123456 |
| |
| mock_chat = Mock() |
| mock_chat.type = "group" |
| mock_chat.id = -333333 # 在允许列表中 |
| |
| mock_update = Mock() |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| result = await message_handler._is_authorized(mock_update) |
| assert result is True |
| |
| @pytest.mark.asyncio |
| async def test_is_authorized_group_chat_unauthorized(self, message_handler): |
| """测试未授权群组""" |
| mock_user = Mock() |
| mock_user.id = 123456 |
| |
| mock_chat = Mock() |
| mock_chat.type = "supergroup" |
| mock_chat.id = -999999 # 不在允许列表中 |
| |
| mock_update = Mock() |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| result = await message_handler._is_authorized(mock_update) |
| assert result is False |
| |
| @pytest.mark.asyncio |
| async def test_is_authorized_unknown_chat_type(self, message_handler): |
| """测试未知聊天类型""" |
| mock_user = Mock() |
| mock_user.id = 123456 |
| |
| mock_chat = Mock() |
| mock_chat.type = "channel" # 未处理的类型 |
| mock_chat.id = -123456 |
| |
| mock_update = Mock() |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| result = await message_handler._is_authorized(mock_update) |
| assert result is False |
| |
| def test_extract_reply_info_with_text(self, message_handler): |
| """测试提取文本回复信息""" |
| # 创建回复的消息 |
| mock_reply_user = Mock(spec=User) |
| mock_reply_user.id = 12345 |
| mock_reply_user.username = "replied_user" |
| mock_reply_user.first_name = "Replied" |
| mock_reply_user.last_name = None |
| mock_reply_user.is_bot = False |
| |
| mock_reply_message = Mock(spec=Message) |
| mock_reply_message.from_user = mock_reply_user |
| mock_reply_message.text = "This is the replied message" |
| mock_reply_message.photo = None |
| mock_reply_message.document = None |
| mock_reply_message.message_id = 100 |
| # Set date to None to avoid mock astimezone() calls |
| mock_reply_message.configure_mock(date=None) |
| |
| # 创建当前消息 |
| mock_message = Mock(spec=Message) |
| mock_message.reply_to_message = mock_reply_message |
| |
| reply_info = message_handler._extract_reply_info(mock_message) |
| |
| expected_reply_info = { |
| 'user_info': { |
| 'user_id': 12345, |
| 'username': "replied_user", |
| 'first_name': "Replied", |
| 'last_name': None, |
| 'is_bot': False |
| }, |
| 'timestamp': None, # No date set on mock |
| 'content': "This is the replied message" |
| } |
| |
| assert reply_info == expected_reply_info |
| |
| def test_extract_reply_info_with_photo(self, message_handler): |
| """测试提取图片回复信息""" |
| mock_reply_user = Mock(spec=User) |
| mock_reply_user.id = 12346 |
| mock_reply_user.username = "photo_user" |
| mock_reply_user.first_name = "Photo" |
| mock_reply_user.last_name = None |
| mock_reply_user.is_bot = False |
| |
| mock_reply_message = Mock(spec=Message) |
| mock_reply_message.from_user = mock_reply_user |
| mock_reply_message.text = None |
| mock_reply_message.photo = [Mock()] # 有图片 |
| mock_reply_message.document = None |
| mock_reply_message.caption = "A beautiful photo" |
| mock_reply_message.message_id = 200 |
| # Set date to None to avoid mock astimezone() calls |
| mock_reply_message.configure_mock(date=None) |
| |
| mock_message = Mock(spec=Message) |
| mock_message.reply_to_message = mock_reply_message |
| |
| reply_info = message_handler._extract_reply_info(mock_message) |
| |
| expected_reply_info = { |
| 'user_info': { |
| 'user_id': 12346, |
| 'username': "photo_user", |
| 'first_name': "Photo", |
| 'last_name': None, |
| 'is_bot': False |
| }, |
| 'timestamp': None, # No date set on mock |
| 'content': "A beautiful photo" |
| } |
| |
| assert reply_info == expected_reply_info |
| |
| def test_extract_reply_info_with_document(self, message_handler): |
| """测试提取文档回复信息""" |
| mock_reply_user = Mock(spec=User) |
| mock_reply_user.id = 12347 |
| mock_reply_user.username = "doc_user" |
| mock_reply_user.first_name = "Document" |
| mock_reply_user.last_name = None |
| mock_reply_user.is_bot = False |
| |
| mock_document = Mock() |
| mock_document.file_name = "test_file.pdf" |
| |
| mock_reply_message = Mock(spec=Message) |
| mock_reply_message.from_user = mock_reply_user |
| mock_reply_message.text = None |
| mock_reply_message.photo = None |
| mock_reply_message.document = mock_document |
| mock_reply_message.caption = "Important document" |
| mock_reply_message.message_id = 300 |
| # Set date to None to avoid mock astimezone() calls |
| mock_reply_message.configure_mock(date=None) |
| |
| mock_message = Mock(spec=Message) |
| mock_message.reply_to_message = mock_reply_message |
| |
| reply_info = message_handler._extract_reply_info(mock_message) |
| |
| expected_reply_info = { |
| 'user_info': { |
| 'user_id': 12347, |
| 'username': "doc_user", |
| 'first_name': "Document", |
| 'last_name': None, |
| 'is_bot': False |
| }, |
| 'timestamp': None, # No date set on mock |
| 'content': "Important document" |
| } |
| |
| assert reply_info == expected_reply_info |
| |
| def test_extract_reply_info_no_caption(self, message_handler): |
| """测试提取无说明的回复信息""" |
| mock_reply_user = Mock(spec=User) |
| mock_reply_user.id = 12348 |
| mock_reply_user.username = "user" |
| mock_reply_user.first_name = "User" |
| mock_reply_user.last_name = None |
| mock_reply_user.is_bot = False |
| |
| mock_reply_message = Mock(spec=Message) |
| mock_reply_message.from_user = mock_reply_user |
| mock_reply_message.text = None |
| mock_reply_message.photo = [Mock()] |
| mock_reply_message.document = None |
| mock_reply_message.caption = None # 无说明 |
| mock_reply_message.message_id = 400 |
| mock_reply_message.date = None # No date |
| |
| mock_message = Mock(spec=Message) |
| mock_message.reply_to_message = mock_reply_message |
| |
| reply_info = message_handler._extract_reply_info(mock_message) |
| |
| # When there's no caption but there are photos, the method returns "[图片]" |
| assert reply_info['content'] == "[图片]" |
| |
| @pytest.mark.asyncio |
| async def test_get_reply_info_private_chat_authorized(self, message_handler): |
| """测试私聊授权用户的回复信息""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 111111 # 授权用户 |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.type = "private" |
| mock_chat.id = 111111 |
| |
| mock_update = Mock(spec=Update) |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| should_reply, is_random = await message_handler._get_reply_info(mock_update, "Hello") |
| |
| assert should_reply is True |
| assert is_random is False |
| |
| @pytest.mark.asyncio |
| async def test_get_reply_info_private_chat_unauthorized(self, message_handler): |
| """测试私聊未授权用户的回复信息""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 999999 # 未授权用户 |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.type = "private" |
| mock_chat.id = 999999 |
| |
| mock_update = Mock(spec=Update) |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| should_reply, is_random = await message_handler._get_reply_info(mock_update, "Hello") |
| |
| assert should_reply is False |
| assert is_random is False |
| |
| @pytest.mark.asyncio |
| async def test_get_reply_info_group_mentioned(self, message_handler): |
| """测试群聊中被提及的情况""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 123456 |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.type = "group" |
| mock_chat.id = -333333 # 授权群组 |
| |
| mock_update = Mock(spec=Update) |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| with patch.object(message_handler.participation_manager, 'is_name_mentioned', return_value=True): |
| should_reply, is_random = await message_handler._get_reply_info(mock_update, "Hello Claude") |
| |
| assert should_reply is True |
| assert is_random is False |
| |
| @pytest.mark.asyncio |
| async def test_get_reply_info_group_random_participation(self, message_handler): |
| """测试群聊随机参与""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 123456 |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.type = "supergroup" |
| mock_chat.id = -444444 # 授权群组 |
| |
| mock_update = Mock(spec=Update) |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| with patch.object(message_handler.participation_manager, 'is_name_mentioned', return_value=False): |
| with patch.object(message_handler.participation_manager, 'should_participate_random', return_value=True): |
| should_reply, is_random = await message_handler._get_reply_info(mock_update, "Normal message") |
| |
| assert should_reply is True |
| assert is_random is True |
| |
| @pytest.mark.asyncio |
| async def test_get_reply_info_group_no_participation(self, message_handler): |
| """测试群聊不参与""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 123456 |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.type = "group" |
| mock_chat.id = -333333 |
| |
| mock_update = Mock(spec=Update) |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| with patch.object(message_handler.participation_manager, 'is_name_mentioned', return_value=False): |
| with patch.object(message_handler.participation_manager, 'should_participate_random', return_value=False): |
| should_reply, is_random = await message_handler._get_reply_info(mock_update, "Normal message") |
| |
| assert should_reply is False |
| assert is_random is False |
| |
| @pytest.mark.asyncio |
| async def test_get_reply_info_unauthorized_group(self, message_handler): |
| """测试未授权群组""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 123456 |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.type = "group" |
| mock_chat.id = -999999 # 未授权群组 |
| |
| mock_update = Mock(spec=Update) |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| should_reply, is_random = await message_handler._get_reply_info(mock_update, "Hello Claude") |
| |
| assert should_reply is False |
| assert is_random is False |
| |
| |
| class TestMessageHandlerEdgeCases: |
| """消息处理器边界情况测试""" |
| |
| @pytest.fixture |
| def minimal_handler(self): |
| """创建最小配置的消息处理器""" |
| mock_components = { |
| 'telegram_client': Mock(), |
| 'context_manager': Mock(), |
| 'file_handler': Mock(), |
| 'claude_agent': Mock(), |
| 'stream_sender': Mock() |
| } |
| |
| return MessageHandler( |
| allowed_users=[], |
| allowed_groups=[], |
| **mock_components |
| ) |
| |
| @pytest.mark.asyncio |
| async def test_empty_allowed_lists(self, minimal_handler): |
| """测试空的授权列表""" |
| mock_user = Mock() |
| mock_user.id = 123456 |
| |
| mock_chat = Mock() |
| mock_chat.type = "private" |
| mock_chat.id = 123456 |
| |
| mock_update = Mock() |
| mock_update.effective_user = mock_user |
| mock_update.effective_chat = mock_chat |
| |
| result = await minimal_handler._is_authorized(mock_update) |
| assert result is False |
| |
| def test_participation_manager_with_empty_config(self): |
| """测试空配置的参与管理器""" |
| manager = GroupParticipationManager(bot_names=[], participation_range=[1, 1]) |
| |
| # 空的bot名字列表 |
| result = manager.is_name_mentioned("Claude help me") |
| assert result is False |
| |
| # 范围为[1,1]意味着每次都会参与 |
| chat_id = "-123" |
| manager.reset_counter(chat_id) |
| result = manager.should_participate_random(chat_id) |
| assert result is True |
| |
| def test_extract_reply_info_no_user(self, minimal_handler): |
| """测试回复消息没有用户信息的情况""" |
| mock_reply_message = Mock(spec=Message) |
| mock_reply_message.from_user = None # 没有用户信息 |
| mock_reply_message.text = "Message without user" |
| mock_reply_message.photo = None |
| mock_reply_message.document = None |
| mock_reply_message.message_id = 500 |
| |
| mock_message = Mock(spec=Message) |
| mock_message.reply_to_message = mock_reply_message |
| |
| reply_info = minimal_handler._extract_reply_info(mock_message) |
| |
| # 应该仍然能提取信息，但用户信息包含默认值 |
| expected_user_info = { |
| 'user_id': 'unknown', |
| 'username': None, |
| 'first_name': 'system', |
| 'last_name': None, |
| 'is_bot': False |
| } |
| assert reply_info['user_info'] == expected_user_info |
| assert reply_info['content'] == "Message without user" |
| |
| def test_build_user_info_extreme_values(self, minimal_handler): |
| """测试极值情况的用户信息构建""" |
| mock_user = Mock(spec=User) |
| mock_user.id = 2**63 - 1 # 最大整数 |
| mock_user.username = "a" * 100 # 很长的用户名 |
| mock_user.first_name = "" # 空字符串 |
| mock_user.last_name = " " # 空白字符串 |
| mock_user.is_bot = True |
| |
| mock_chat = Mock(spec=Chat) |
| mock_chat.id = -(2**63) # 最小整数 |
| mock_chat.title = "群组" * 100 # 很长的群组名 |
| mock_chat.type = "supergroup" |
| |
| user_info = minimal_handler._build_user_info(mock_user, mock_chat) |
| |
| # 验证所有值都被正确处理 |
| assert user_info['user_id'] == 2**63 - 1 |
| assert user_info['username'] == "a" * 100 |
| assert user_info['first_name'] == "" |
| assert user_info['last_name'] == " " |
| # The actual _build_user_info method doesn't include is_bot field |
| # assert user_info['is_bot'] is True # This field is not included in actual implementation |
| assert user_info['chat_id'] == -(2**63) |
| assert user_info['chat_title'] == "群组" * 100 |
| assert user_info['chat_type'] == "supergroup" |
| |
| |
| class TestBotInteractionFunctionality: |
| """Bot交互功能测试（从test_bot_interaction_simple.py合并）""" |
| |
| @pytest.fixture |
| def mock_components(self): |
| """创建模拟组件""" |
| return { |
| 'telegram_client': Mock(), |
| 'context_manager': Mock(), |
| 'file_handler': Mock(), |
| 'claude_agent': Mock(), |
| 'stream_sender': Mock() |
| } |
| |
| @pytest.fixture |
| def message_handler(self, mock_components): |
| """创建消息处理器""" |
| return MessageHandler( |
| telegram_client=mock_components['telegram_client'], |
| context_manager=mock_components['context_manager'], |
| file_handler=mock_components['file_handler'], |
| claude_agent=mock_components['claude_agent'], |
| stream_sender=mock_components['stream_sender'], |
| allowed_users=[123456], |
| allowed_groups=[789012] |
| ) |
| |
| def test_message_handler_initialization_with_lists(self, message_handler): |
| """测试消息处理器初始化（使用列表参数）""" |
| assert message_handler.allowed_users == {123456} |
| assert message_handler.allowed_groups == {789012} |
| assert message_handler.participation_manager is not None |
| |
| def test_bot_mention_detection_webhook(self, message_handler): |
| """测试Webhook Bot提及检测""" |
| # 设置bot用户名 |
| message_handler.bot_username = "TestBot" |
| |
| # 测试@mention检测 |
| result = message_handler._check_mention_in_text_webhook("Hello @TestBot, how are you?") |
| assert result is True |
| |
| # 测试无@mention |
| result = message_handler._check_mention_in_text_webhook("Hello everyone!") |
| assert result is False |
| |
| def test_build_user_info_comprehensive(self, message_handler): |
| """测试构建用户信息（全面测试）""" |
| from telegram import User, Chat |
| |
| user = Mock(spec=User) |
| user.id = 123456 |
| user.username = "testuser" |
| user.first_name = "Test" |
| user.last_name = "User" |
| user.language_code = "en" |
| |
| chat = Mock(spec=Chat) |
| chat.id = 789012 |
| chat.type = "group" |
| chat.title = "Test Group" |
| |
| user_info = message_handler._build_user_info(user, chat) |
| |
| assert user_info['user_id'] == 123456 |
| assert user_info['username'] == "testuser" |
| assert user_info['first_name'] == "Test" |
| assert user_info['chat_id'] == 789012 |
| assert user_info['chat_type'] == "group" |
| |
| def test_format_reply_context_user(self, message_handler): |
| """测试回复上下文格式化（用户）""" |
| from datetime import datetime |
| |
| reply_info = { |
| 'user_info': { |
| 'user_id': 999, |
| 'username': 'testuser', |
| 'first_name': 'Test', |
| 'is_bot': False |
| }, |
| 'timestamp': datetime(2025, 10, 4, 10, 30), |
| 'content': 'Hello world' |
| } |
| |
| result = message_handler._format_reply_context(reply_info) |
| expected = "↳ 回复 [@testuser 10:30]: Hello world" |
| assert result == expected |
| |
| def test_format_reply_context_bot_user(self, message_handler): |
| """测试Bot回复上下文格式化""" |
| from datetime import datetime |
| |
| reply_info = { |
| 'user_info': { |
| 'user_id': 888, |
| 'username': 'testbot', |
| 'first_name': 'TestBot', |
| 'is_bot': True |
| }, |
| 'timestamp': datetime(2025, 10, 4, 10, 30), |
| 'content': 'Bot response' |
| } |
| |
| result = message_handler._format_reply_context(reply_info) |
| expected = "↳ 回复 [🤖 @testbot 10:30]: Bot response" |
| assert result == expected |
| |
| def test_build_user_display_name_variations(self, message_handler): |
| """测试构建用户显示名（各种情况）""" |
| # 有用户名的情况 |
| user_info = {'username': 'testuser'} |
| result = message_handler._build_user_display_name(123, user_info) |
| assert result == "@testuser" |
| |
| # 只有名字的情况 |
| user_info = {'first_name': 'Test', 'last_name': 'User'} |
| result = message_handler._build_user_display_name(123, user_info) |
| assert result == "Test User" |
| |
| # 只有first_name的情况 |
| user_info = {'first_name': 'Test'} |
| result = message_handler._build_user_display_name(123, user_info) |
| assert result == "Test" |
| |
| # 没有任何信息的情况 |
| user_info = {} |
| result = message_handler._build_user_display_name(123, user_info) |
| assert result == "User123" |