test: 添加Celery邮件测试和日志配置检查功能

添加两个新的测试用例： 1. test_celery_redis_email：测试通过Celery和Redis发送邮件功能 2. check_logs_config：检查Gunicorn、Celery、Redis的日志配置是否统一同时更新主测试函数以包含新增测试
2026-01-19 21:06:39 +08:00
parent ce575e6b35
commit e2f389a325
1 changed files with 141 additions and 5 deletions
--- a/test_redis_celery.py
+++ b/test_redis_celery.py
@@ -304,6 +304,125 @@ def test_redis_performance():
        logger.error(f"Redis性能测试失败: {e}")
        return False

+
+def test_celery_redis_email():
+    """测试通过Celery和Redis发送邮件功能"""
+    logger.info("开始测试通过Celery和Redis发送邮件...")
+    try:
+        from core.tasks import celery_send_test_email
+        
+        # 发送测试邮件任务
+        logger.info("发送Celery邮件测试任务到Redis队列...")
+        task = celery_send_test_email.delay(test_mode=True)
+        logger.info(f"邮件测试任务已发送，任务ID: {task.id}")
+        
+        # 等待任务完成
+        logger.info("等待任务执行（最多30秒）...")
+        result = task.get(timeout=30)
+        logger.info(f"邮件任务执行结果: {result}")
+        
+        # 检查结果状态
+        if result.get('status') == 'success':
+            logger.success("通过Celery和Redis发送邮件测试成功！")
+            return True
+        else:
+            logger.error(f"邮件发送失败: {result.get('error', '未知错误')}")
+            return False
+            
+    except Exception as e:
+        error_msg = str(e)
+        logger.error(f"通过Celery和Redis发送邮件测试失败: {error_msg}")
+        
+        # 提供具体的解决方案
+        if "No such file or directory" in error_msg:
+            logger.error("📁 文件路径错误，可能的原因:")
+            logger.error("1. 配置文件路径不存在")
+            logger.error("2. 权限问题导致无法访问文件")
+        elif "Connection refused" in error_msg:
+            logger.error("🔌 连接被拒绝，可能的原因:")
+            logger.error("1. Celery worker未运行")
+            logger.error("2. Redis服务未运行")
+            logger.error("3. 防火墙阻止连接")
+        else:
+            logger.error("请检查:")
+            logger.error("1. Celery worker是否运行: sudo supervisorctl status celery_worker")
+            logger.error("2. Redis服务状态: sudo systemctl status redis-server")
+            logger.error("3. 系统配置中的邮件设置是否正确")
+            logger.error("4. SMTP服务器配置是否正确")
+        
+        return False
+
+
+def check_logs_config():
+    """检查Gunicorn、Celery、Redis的日志是否写入同一个文件"""
+    logger.info("开始检查日志配置...")
+    try:
+        from django.conf import settings
+        
+        # 检查Django日志配置
+        if hasattr(settings, 'LOGGING'):
+            logging_config = settings.LOGGING
+            logger.info("Django日志配置:")
+            
+            # 检查日志处理器
+            handlers = logging_config.get('handlers', {})
+            file_handlers = [name for name, handler in handlers.items() if handler.get('class') == 'logging.handlers.RotatingFileHandler']
+            
+            if file_handlers:
+                for handler_name in file_handlers:
+                    handler = handlers[handler_name]
+                    log_file = handler.get('filename', '未知')
+                    logger.info(f"  文件日志处理器 '{handler_name}' 写入到: {log_file}")
+                    
+                    # 检查日志器配置
+                    loggers = logging_config.get('loggers', {})
+                    logger_names = []
+                    for logger_name, logger_config in loggers.items():
+                        if handler_name in logger_config.get('handlers', []):
+                            logger_names.append(logger_name)
+                    
+                    if logger_names:
+                        logger.info(f"  使用该处理器的日志器: {', '.join(logger_names)}")
+                        
+                        # 检查是否包含celery日志
+                        if 'celery' in logger_names:
+                            logger.info("  ✅ Celery日志将写入该文件")
+                        else:
+                            logger.warning("  ⚠️ Celery日志未配置使用该处理器")
+            else:
+                logger.warning("  未找到文件日志处理器配置")
+        else:
+            logger.warning("  未配置Django LOGGING设置")
+        
+        # 检查Celery日志配置
+        if hasattr(settings, 'CELERY_LOG_FILE'):
+            celery_log_file = settings.CELERY_LOG_FILE
+            logger.info(f"Celery工作进程日志配置: {celery_log_file}")
+        else:
+            logger.warning("未配置CELERY_LOG_FILE")
+        
+        # 检查Gunicorn日志配置（通常在Gunicorn配置文件中，这里检查环境变量或常见配置）
+        logger.info("Gunicorn日志配置:")
+        logger.info("  注意: Gunicorn日志通常在其配置文件或启动命令中设置")
+        logger.info("  建议配置Gunicorn日志写入与Django/Celery相同的日志文件")
+        
+        # 检查Redis日志配置
+        logger.info("Redis日志配置:")
+        logger.info("  注意: Redis日志通常在/etc/redis/redis.conf中配置")
+        logger.info("  建议配置Redis日志写入与其他服务相同的日志目录")
+        
+        logger.info("\n日志配置检查完成！")
+        logger.info("建议:")
+        logger.info("1. 确保Django、Celery、Gunicorn、Redis的日志都写入同一个文件")
+        logger.info("2. 配置日志轮转和保留策略，避免日志文件过大")
+        logger.info("3. 定期检查日志文件，及时发现问题")
+        
+        return True
+        
+    except Exception as e:
+        logger.error(f"日志配置检查失败: {e}")
+        return False
+
 def main():
    """主测试函数"""
    logger.info("=== Redis和Celery生产环境配置测试开始 ===")
@@ -311,7 +430,7 @@ def main():
    logger.info("=" * 50)
    
    tests_passed = 0
-    total_tests = 3
+    total_tests = 5
    
    # 测试1: Redis连接
    logger.info("\n[测试1] Redis连接测试")
@@ -330,6 +449,16 @@ def main():
    if test_redis_performance():
        tests_passed += 1
    
+    # 测试4: 通过Celery和Redis发送邮件
+    logger.info("\n[测试4] 通过Celery和Redis发送邮件测试")
+    if test_celery_redis_email():
+        tests_passed += 1
+    
+    # 测试5: 检查日志配置
+    logger.info("\n[测试5] 检查日志配置")
+    if check_logs_config():
+        tests_passed += 1
+    
    # 测试总结
    logger.info("\n" + "=" * 50)
    logger.info("测试总结:")
@@ -341,15 +470,20 @@ def main():
        logger.info("1. ✅ Redis服务运行正常")
        logger.info("2. ✅ Celery可以连接到Redis")
        logger.info("3. ✅ Redis性能满足要求")
-        logger.info("4. 建议配置Redis持久化和备份")
-        logger.info("5. 建议监控Redis内存使用情况")
+        logger.info("4. ✅ 可以通过Celery和Redis发送邮件")
+        logger.info("5. ✅ 日志配置检查完成")
+        logger.info("6. 建议配置Redis持久化和备份")
+        logger.info("7. 建议监控Redis内存使用情况")
+        logger.info("8. 确保Gunicorn、Celery、Redis日志写入同一个文件")
        return 0
-    elif tests_passed >= 2:
+    elif tests_passed >= 3:
        logger.warning("部分测试通过，生产环境基本可用。")
        logger.info("\n需要检查:")
        logger.info("1. 确保Redis服务正常运行")
        logger.info("2. 检查Celery worker配置")
        logger.info("3. 参考README中的故障排除指南")
+        logger.info("4. 检查邮件配置是否正确")
+        logger.info("5. 检查日志配置是否符合要求")
        return 1
    else:
        logger.error("多数测试失败，生产环境可能无法正常工作。")
@@ -357,7 +491,9 @@ def main():
        logger.info("1. ❌ 检查Redis服务状态: sudo systemctl status redis-server")
        logger.info("2. ❌ 检查Redis配置: /etc/redis/redis.conf")
        logger.info("3. ❌ 检查Django settings.py中的Celery配置")
-        logger.info("4. 参考README中的Redis部署章节重新配置")
+        logger.info("4. ❌ 检查Celery worker是否运行")
+        logger.info("5. ❌ 检查系统配置中的邮件设置")
+        logger.info("6. 参考README中的Redis部署章节重新配置")
        return 1

 if __name__ == "__main__":