"""
|
消息处理器测试
|
"""
|
import pytest
|
from unittest.mock import Mock, patch
|
from app.services.message_processor import MessageProcessor
|
|
|
class TestMessageProcessor:
|
"""消息处理器测试类"""
|
|
def setup_method(self):
|
"""测试前准备"""
|
self.processor = MessageProcessor()
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_success(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试有效群聊消息验证"""
|
mock_silence_service.is_silence_active.return_value = False
|
mock_friend_ignore_service.is_friend_ignored.return_value = False
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is True
|
|
def test_is_valid_group_message_wrong_type(self):
|
"""测试错误消息类型"""
|
callback_data = {
|
"messageType": "80002", # 非群聊消息
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
|
def test_is_valid_group_message_self_sent(self):
|
"""测试自己发送的消息"""
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": True # 自己发送的消息
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
|
def test_is_valid_group_message_missing_fields(self):
|
"""测试缺少必要字段"""
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
# 缺少 fromGroup 和 content
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_friend_ignored(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试好友在忽略列表中的消息"""
|
mock_silence_service.is_silence_active.side_effect = [False, False] # 两次检查都返回False
|
mock_silence_service.activate_silence_mode.return_value = True
|
mock_friend_ignore_service.is_friend_ignored.return_value = True
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
mock_friend_ignore_service.is_friend_ignored.assert_called_once_with("wxid_test123")
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_friend_not_ignored(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试好友不在忽略列表中的消息"""
|
mock_silence_service.is_silence_active.return_value = False
|
mock_friend_ignore_service.is_friend_ignored.return_value = False
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is True
|
mock_friend_ignore_service.is_friend_ignored.assert_called_once_with("wxid_test123")
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_silence_active(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试群组静默模式激活时的消息处理（非忽略好友）"""
|
mock_friend_ignore_service.is_friend_ignored.return_value = False
|
mock_silence_service.is_silence_active.return_value = True
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
mock_silence_service.is_silence_active.assert_called_once_with("group123@chatroom")
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_friend_ignored_activate_silence(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试好友被忽略时激活静默模式"""
|
mock_silence_service.is_silence_active.side_effect = [False, False] # 第一次检查未激活，第二次检查忽略好友时也未激活
|
mock_silence_service.activate_silence_mode.return_value = True
|
mock_friend_ignore_service.is_friend_ignored.return_value = True
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
mock_silence_service.activate_silence_mode.assert_called_once_with("group123@chatroom")
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_friend_ignored_extend_silence(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试好友被忽略时延长群组静默模式"""
|
mock_silence_service.is_silence_active.return_value = True # 群组静默模式已激活
|
mock_silence_service.extend_silence_mode.return_value = True
|
mock_friend_ignore_service.is_friend_ignored.return_value = True
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
mock_silence_service.extend_silence_mode.assert_called_once_with("group123@chatroom")
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
def test_is_valid_group_message_friend_ignored_in_silence_mode(self, mock_friend_ignore_service, mock_silence_service):
|
"""测试群组静默模式下好友消息仍能刷新时长"""
|
# 模拟群组静默模式已激活的情况
|
mock_friend_ignore_service.is_friend_ignored.return_value = True
|
mock_silence_service.is_silence_active.return_value = True
|
mock_silence_service.extend_silence_mode.return_value = True
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_ignored_friend",
|
"fromGroup": "group123@chatroom",
|
"content": "被忽略好友在群组静默模式下的消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.is_valid_group_message(callback_data)
|
assert result is False
|
|
# 验证好友忽略检查被调用
|
mock_friend_ignore_service.is_friend_ignored.assert_called_once_with("wxid_ignored_friend")
|
# 验证群组静默模式时间被延长
|
mock_silence_service.extend_silence_mode.assert_called_once_with("group123@chatroom")
|
|
@patch('app.services.message_processor.silence_service')
|
@patch('app.services.message_processor.friend_ignore_service')
|
@patch('app.services.message_processor.redis_queue')
|
def test_enqueue_callback_message_success(self, mock_redis_queue, mock_friend_ignore_service, mock_silence_service):
|
"""测试消息入队成功"""
|
mock_silence_service.is_silence_active.return_value = False
|
mock_friend_ignore_service.is_friend_ignored.return_value = False
|
mock_redis_queue.enqueue_message.return_value = True
|
|
callback_data = {
|
"messageType": "80001",
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.enqueue_callback_message(callback_data)
|
|
assert result is True
|
mock_redis_queue.enqueue_message.assert_called_once_with("wxid_test123", callback_data)
|
|
@patch('app.services.message_processor.redis_queue')
|
def test_enqueue_callback_message_invalid(self, mock_redis_queue):
|
"""测试无效消息入队"""
|
callback_data = {
|
"messageType": "80002", # 无效消息类型
|
"data": {
|
"fromUser": "wxid_test123",
|
"fromGroup": "group123@chatroom",
|
"content": "测试消息",
|
"self": False
|
}
|
}
|
|
result = self.processor.enqueue_callback_message(callback_data)
|
|
assert result is False
|
mock_redis_queue.enqueue_message.assert_not_called()
|
|
|
if __name__ == "__main__":
|
pytest.main([__file__])
|
