使用pytest测试Celery任务时出现永久挂起问题求助

1. 修复apply_async参数传递错误 #

你的测试代码中read_dynamodb.apply_async("us-west-2", "blank")的参数传递方式错误。apply_async的第一个位置参数是args，需要传入元组格式的任务参数，否则会被解析为args="us-west-2", kwargs="blank"，导致任务执行时参数不匹配，worker抛出异常但未正确写入结果到DynamoDB，最终result.get()无限等待。

修正后的代码：

def test_read_dynamodb(self):
    # 正确传递参数为元组
    result = read_dynamodb.apply_async(("us-west-2", "blank"))
    # 或者显式指定args参数
    # result = read_dynamodb.apply_async(args=("us-west-2", "blank"))
    print(result.get(timeout=10))  # 建议添加超时避免永久挂起

2. 为result.get()添加超时机制 #

即使参数正确，也建议为get()方法设置超时时间，避免因网络或后端问题导致测试永久挂起：

result.get(timeout=10)  # 设置10秒超时，根据任务实际耗时调整

3. 验证pytest-celery worker的配置与状态 #

• 确保celery_session_worker fixture正确启动了worker，并且worker能连接到SQS和DynamoDB。可以在测试中添加检查：

  def test_read_dynamodb(self, celery_session_worker):
      print(celery_session_worker.is_alive())  # 检查worker是否存活
      result = read_dynamodb.apply_async(("us-west-2", "blank"))
      print(result.get(timeout=10))
  

• 开启worker日志，确认任务是否被接收和执行：

  # tests/conftest.py
  @pytest.fixture(scope="session")
  def celery_config():
      return {
          "broker": "sqs://",
          "broker_transport_options": broker_transport_options,
          "backend": "dynamodb://@us-east-2/awesome_queue",
          "worker_log_level": "INFO",  # 开启worker日志
      }
  

4. 检查DynamoDB权限与表配置 #

• 确保pytest运行时使用的AWS凭证有DynamoDB的读写权限（dynamodb:PutItem、dynamodb:GetItem等）。
• 确认DynamoDB表awesome_queue的主键为task_id（Celery DynamoDB后端默认主键），表结构符合要求。

5. 排查M1芯片兼容性问题 #

在Apple M1芯片上，部分Python库可能存在兼容性问题：

• 尝试切换到Python 3.10版本（Celery 5.2.7对Python 3.11的适配可能存在问题）。
• 确保boto3、celery[sqs]等依赖包为ARM架构编译版本。

内容的提问来源于stack exchange，提问作者Ken