97 lines
2.8 KiB
Python
97 lines
2.8 KiB
Python
"""
|
|
消息队列单元测试
|
|
"""
|
|
import pytest
|
|
from app.services.message_queue import MessageQueue
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
|
|
class TestMessageQueue:
|
|
"""消息队列测试"""
|
|
|
|
@pytest.fixture
|
|
def mock_redis(self):
|
|
"""模拟 Redis 客户端"""
|
|
with patch('app.services.message_queue.redis_client') as mock:
|
|
mock.hset.return_value = 1
|
|
mock.rpush.return_value = 1
|
|
mock.lpop.return_value = None
|
|
mock.hgetall.return_value = {}
|
|
mock.delete.return_value = 1
|
|
mock.llen.return_value = 0
|
|
yield mock
|
|
|
|
def test_enqueue(self, mock_redis):
|
|
"""测试消息入队"""
|
|
message = MagicMock()
|
|
message.id = 'msg-123'
|
|
message.session_id = 'session-123'
|
|
message.sender_type = 'user'
|
|
message.sender_id = 'user-123'
|
|
message.content = 'Hello'
|
|
message.status = 'sent'
|
|
message.retry_count = 0
|
|
message.created_at = None
|
|
|
|
result = MessageQueue.enqueue(message)
|
|
|
|
assert result is True
|
|
mock_redis.hset.assert_called()
|
|
mock_redis.rpush.assert_called()
|
|
|
|
def test_dequeue_empty(self, mock_redis):
|
|
"""测试空队列出队"""
|
|
mock_redis.lpop.return_value = None
|
|
|
|
result = MessageQueue.dequeue()
|
|
|
|
assert result is None
|
|
|
|
def test_dequeue_success(self, mock_redis):
|
|
"""测试成功出队"""
|
|
mock_redis.lpop.return_value = 'msg-123'
|
|
mock_redis.hgetall.return_value = {
|
|
'id': 'msg-123',
|
|
'session_id': 'session-123',
|
|
'content': 'Hello',
|
|
'retry_count': '0',
|
|
}
|
|
|
|
result = MessageQueue.dequeue()
|
|
|
|
assert result is not None
|
|
assert result['id'] == 'msg-123'
|
|
|
|
def test_ack(self, mock_redis):
|
|
"""测试消息确认"""
|
|
result = MessageQueue.ack('msg-123')
|
|
|
|
assert result is True
|
|
mock_redis.delete.assert_called()
|
|
|
|
def test_retry_within_limit(self, mock_redis):
|
|
"""测试重试（未超限）"""
|
|
mock_redis.hget.return_value = '1'
|
|
|
|
result = MessageQueue.retry('msg-123')
|
|
|
|
assert result is True
|
|
mock_redis.hset.assert_called()
|
|
mock_redis.rpush.assert_called()
|
|
|
|
def test_retry_exceed_limit(self, mock_redis):
|
|
"""测试重试（超限）"""
|
|
mock_redis.hget.return_value = '3'
|
|
|
|
result = MessageQueue.retry('msg-123')
|
|
|
|
assert result is False
|
|
|
|
def test_get_pending_count(self, mock_redis):
|
|
"""测试获取待处理数量"""
|
|
mock_redis.llen.return_value = 5
|
|
|
|
count = MessageQueue.get_pending_count()
|
|
|
|
assert count == 5
|