"""
|
消息处理器测试
|
"""
|
import pytest
|
from unittest.mock import Mock, patch
|
from app.services.message_processor import MessageProcessor
|
|
|
class TestMessageProcessor:
|
"""消息处理器测试类"""
|
|
def setup_method(self):
|
"""测试前准备"""
|
self.processor = MessageProcessor()
|
|
def test_is_valid_group_message_success(self):
|
"""测试有效群聊消息验证"""
|
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.redis_queue')
|
def test_enqueue_callback_message_success(self, mock_redis_queue):
|
"""测试消息入队成功"""
|
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__])
|
