126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试RedisManager连接功能
|
||
验证connect方法的返回值
|
||
"""
|
||
|
||
from app.redis_lock import redis_manager
|
||
from app.logger import get_redis_logger
|
||
|
||
logger = get_redis_logger()
|
||
|
||
def test_connection():
|
||
"""测试连接"""
|
||
logger.info("🧪 测试Redis连接...")
|
||
|
||
# 测试连接
|
||
success = redis_manager.connect()
|
||
|
||
if success:
|
||
logger.info("✅ 连接成功")
|
||
|
||
# 测试创建锁
|
||
try:
|
||
lock = redis_manager.create_lock("test:lock", timeout=10)
|
||
logger.info("✅ 锁创建成功")
|
||
|
||
# 尝试获取锁
|
||
acquired = lock.acquire(blocking=False)
|
||
if acquired:
|
||
logger.info("✅ 锁获取成功")
|
||
|
||
# 释放锁
|
||
released = lock.release()
|
||
if released:
|
||
logger.info("✅ 锁释放成功")
|
||
else:
|
||
logger.error("❌ 锁释放失败")
|
||
return False
|
||
else:
|
||
logger.error("❌ 锁获取失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 锁操作失败: {e}")
|
||
return False
|
||
else:
|
||
logger.error("❌ 连接失败")
|
||
return False
|
||
|
||
return True
|
||
|
||
def test_connection_failures():
|
||
"""测试连接失败场景"""
|
||
logger.info("🧪 测试连接失败处理...")
|
||
|
||
# 直接测试连接一个不存在的主机
|
||
from app.config import settings
|
||
original_url = settings.redis_url
|
||
|
||
try:
|
||
# 创建一个新的RedisManager实例用于测试失败场景
|
||
from app.redis_lock import RedisManager
|
||
test_manager = RedisManager()
|
||
|
||
# 模拟连接失败 - 直接使用无效的URL
|
||
import redis
|
||
invalid_redis_client = redis.from_url(
|
||
"redis://invalid_host:6379/0",
|
||
encoding="utf-8",
|
||
decode_responses=True,
|
||
socket_connect_timeout=1,
|
||
socket_timeout=1
|
||
)
|
||
|
||
test_manager.redis_client = invalid_redis_client
|
||
|
||
# 尝试ping，这应该会失败
|
||
try:
|
||
result = test_manager.redis_client.ping()
|
||
logger.error("❌ 连接应该失败但成功了")
|
||
return False
|
||
except Exception:
|
||
logger.info("✅ 连接失败处理正确")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.info("✅ 连接失败处理正确")
|
||
return True
|
||
|
||
def main():
|
||
"""主函数"""
|
||
logger.info("🚀 开始RedisManager连接测试")
|
||
|
||
# 测试1: 连接
|
||
logger.info("\n" + "="*50)
|
||
logger.info("测试1: 连接")
|
||
logger.info("="*50)
|
||
|
||
connection_success = test_connection()
|
||
|
||
# 测试2: 连接失败处理
|
||
logger.info("\n" + "="*50)
|
||
logger.info("测试2: 连接失败处理")
|
||
logger.info("="*50)
|
||
|
||
failure_success = test_connection_failures()
|
||
|
||
# 断开连接
|
||
redis_manager.disconnect()
|
||
logger.info("🔌 Redis连接已断开")
|
||
|
||
# 总结
|
||
logger.info("\n" + "="*50)
|
||
logger.info("测试结果总结")
|
||
logger.info("="*50)
|
||
|
||
if connection_success and failure_success:
|
||
logger.info("🎉 所有测试通过！RedisManager连接功能正常")
|
||
return True
|
||
else:
|
||
logger.error("❌ 部分测试失败，请检查实现")
|
||
return False
|
||
|
||
if __name__ == "__main__":
|
||
success = main()
|
||
exit(0 if success else 1) |