105 lines
2.9 KiB
Python
105 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Redis 连接测试脚本
|
|
"""
|
|
import os
|
|
import sys
|
|
import asyncio
|
|
|
|
# 添加项目根目录到Python路径
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from app.config import settings
|
|
from app.logger import get_logger
|
|
|
|
logger = get_logger("redis_test")
|
|
|
|
async def test_redis_connection():
|
|
"""测试 Redis 连接"""
|
|
logger.info("🔴 测试 Redis 连接...")
|
|
|
|
try:
|
|
import redis.asyncio as redis
|
|
|
|
# 创建 Redis 客户端
|
|
redis_client = redis.from_url(settings.celery_broker_url)
|
|
|
|
# 测试连接
|
|
await redis_client.ping()
|
|
|
|
logger.info("✅ Redis 连接成功")
|
|
|
|
# 测试基本操作
|
|
test_key = "test_connection"
|
|
test_value = "Hello Redis!"
|
|
|
|
await redis_client.set(test_key, test_value, ex=60) # 设置60秒过期
|
|
retrieved_value = await redis_client.get(test_key)
|
|
|
|
if retrieved_value == test_value:
|
|
logger.info("✅ Redis 读写操作正常")
|
|
else:
|
|
logger.warning("⚠️ Redis 读写操作异常")
|
|
|
|
# 清理测试数据
|
|
await redis_client.delete(test_key)
|
|
|
|
# 关闭连接
|
|
await redis_client.close()
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Redis 连接失败: {e}")
|
|
return False
|
|
|
|
async def test_celery_redis():
|
|
"""测试 Celery 相关的 Redis 操作"""
|
|
logger.info("🌿 测试 Celery Redis 配置...")
|
|
|
|
try:
|
|
import redis.asyncio as redis
|
|
|
|
# 测试 Broker 连接
|
|
broker_client = redis.from_url(settings.celery_broker_url)
|
|
await broker_client.ping()
|
|
logger.info(f"✅ Celery Broker (Redis) 连接成功: {settings.celery_broker_url}")
|
|
|
|
# 测试 Backend 连接
|
|
backend_client = redis.from_url(settings.celery_result_backend)
|
|
await backend_client.ping()
|
|
logger.info(f"✅ Celery Backend (Redis) 连接成功: {settings.celery_result_backend}")
|
|
|
|
# 关闭连接
|
|
await broker_client.close()
|
|
await backend_client.close()
|
|
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"❌ Celery Redis 测试失败: {e}")
|
|
return False
|
|
|
|
async def main():
|
|
"""主函数"""
|
|
logger.info("🧪 开始 Redis 连接测试...")
|
|
|
|
# 显示配置信息
|
|
logger.info(f"📡 Celery Broker: {settings.celery_broker_url}")
|
|
logger.info(f"💾 Celery Backend: {settings.celery_result_backend}")
|
|
|
|
# 测试基础连接
|
|
basic_ok = await test_redis_connection()
|
|
|
|
# 测试 Celery 连接
|
|
celery_ok = await test_celery_redis()
|
|
|
|
if basic_ok and celery_ok:
|
|
logger.info("🎉 所有 Redis 测试通过！")
|
|
return 0
|
|
else:
|
|
logger.error("❌ Redis 测试失败")
|
|
return 1
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(asyncio.run(main())) |