K2CRO4
67d01392e4
Some checks failed
Auto Deploy NeoBot (FRP + SSH 密码登录) / deploy-to-server (push) Has been cancelled
1. 新增反馈插件、复读插件、戳一戳插件 2. 修复了配置、线程安全、SQL校验等多处bug 3. 重构插件加载系统，支持验证插件+热加载 4. 修复大量测试用例问题，修复了76个测试挂逼的问题 5. 调整了broadcast插件的发送间隔 6. 优化了性能统计的函数命名逻辑 7. 修复了furry插件的注释和函数名错误 8. 重构了输入校验的逻辑顺序 9. 配置文件新增了默认值处理
100 lines
3.7 KiB
Python
100 lines
3.7 KiB
Python
import pytest
|
|
from unittest.mock import patch, AsyncMock
|
|
from neobot.core.managers.redis_manager import RedisManager
|
|
|
|
|
|
class TestRedisManager:
|
|
def test_singleton_pattern(self):
|
|
"""测试单例模式。"""
|
|
instance1 = RedisManager()
|
|
instance2 = RedisManager()
|
|
assert instance1 is instance2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_success(self):
|
|
"""测试 Redis 初始化成功。"""
|
|
# 重置 Singleton 状态
|
|
RedisManager._redis = None
|
|
manager = RedisManager()
|
|
if '_redis' in manager.__dict__:
|
|
del manager.__dict__['_redis']
|
|
|
|
with patch('neobot.core.managers.redis_manager.config') as mock_config:
|
|
mock_config.redis.host = "localhost"
|
|
mock_config.redis.port = 6379
|
|
mock_config.redis.db = 0
|
|
mock_config.redis.password = "test_password"
|
|
|
|
with patch('neobot.core.managers.redis_manager.redis.Redis') as mock_redis_class:
|
|
mock_redis = AsyncMock()
|
|
mock_redis.ping.return_value = True
|
|
mock_redis_class.return_value = mock_redis
|
|
|
|
await manager.initialize()
|
|
|
|
mock_redis_class.assert_called_once()
|
|
mock_redis.ping.assert_called_once()
|
|
assert manager._redis is not None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initialize_connection_error(self):
|
|
"""测试 Redis 连接失败。"""
|
|
RedisManager._redis = None
|
|
manager = RedisManager()
|
|
if '_redis' in manager.__dict__:
|
|
del manager.__dict__['_redis']
|
|
|
|
with patch('neobot.core.managers.redis_manager.config') as mock_config:
|
|
mock_config.redis.host = "localhost"
|
|
mock_config.redis.port = 6379
|
|
mock_config.redis.db = 0
|
|
mock_config.redis.password = "test_password"
|
|
|
|
with patch('neobot.core.managers.redis_manager.redis.Redis') as mock_redis_class:
|
|
mock_redis_class.side_effect = Exception("Connection refused")
|
|
|
|
await manager.initialize()
|
|
|
|
assert manager._redis is None
|
|
|
|
def test_redis_property_uninitialized(self):
|
|
"""测试 Redis 属性在未初始化时抛出异常。"""
|
|
RedisManager._redis = None
|
|
manager = RedisManager()
|
|
if '_redis' in manager.__dict__:
|
|
del manager.__dict__['_redis']
|
|
|
|
with pytest.raises(ConnectionError, match="Redis 未初始化或连接失败，请先调用 initialize()"):
|
|
_ = manager.redis
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_method(self):
|
|
"""测试 get 方法。"""
|
|
RedisManager._redis = None
|
|
manager = RedisManager()
|
|
if '_redis' in manager.__dict__:
|
|
del manager.__dict__['_redis']
|
|
|
|
mock_redis = AsyncMock()
|
|
mock_redis.get.return_value = "test_value"
|
|
manager._redis = mock_redis
|
|
|
|
result = await manager.get("test_key")
|
|
assert result == "test_value"
|
|
mock_redis.get.assert_called_once_with("test_key")
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_set_method(self):
|
|
"""测试 set 方法。"""
|
|
RedisManager._redis = None
|
|
manager = RedisManager()
|
|
if '_redis' in manager.__dict__:
|
|
del manager.__dict__['_redis']
|
|
|
|
mock_redis = AsyncMock()
|
|
mock_redis.set.return_value = True
|
|
manager._redis = mock_redis
|
|
|
|
result = await manager.set("test_key", "test_value", ex=3600)
|
|
assert result is True
|
|
mock_redis.set.assert_called_once_with("test_key", "test_value", ex=3600) |