Skip to content

对Celery进行测试

Celery是一个分布式任务队列，它可以让你的任务异步执行。在TFRobotServer中，我们使用Celery来执行一些异步任务，比如定时任务（刷新AccessToken）、后台任务（partition_doc）等。

目前Celery文档中，关于部分的描述不够清晰。结合互联网上的一些内容与源码分析，目前我们初步完成了Celery单元与集成测试的规范制定与实现。主要实现思路如下，如果在开发过程中遇到Celery测试相关问题，可以参数这篇文档。

Celery测试框架介绍

目前Celery提供了两个测试框架：

  1. celery.contrib.pytest
  2. pytest-celery

两个框架的实现思路与使用方法有非常大的不同。对于celery.contrib.pytest使用的是使用独立进程的方式启动一个Worker进行测试。而pytest-celery是使用Docker的方式启动一个Worker进行测试。

对于开发过程中，我们先使用celery.contrib.pytest框架进行测试。原因有以下几点：

  1. 运行速度快。因为避免了Docker的启停时间，所以运行速度会快很多，也进一步方便测试。
  2. 调试方便。因为是直接启动Worker，所以可以直接在Worker中打断点进行调试。
  3. 依赖少。因为不需要Docker，所以依赖少，也方便在CI中使用。
  4. 对于Celery如果不是特别熟悉，使用celery.contrib.pytest框架可以更快的上手。

celery.contrib.pytest使用介绍

celery.contrib.pytest是Celery官方提供的测试框架，使用文档参考。

这里需要注意的是，文档中提到可以通过

add an environment variable PYTEST_PLUGINS=celery.contrib.pytest

or add pytest_plugins = ("celery.contrib.pytest", ) to your root conftest.py

的方式启用celery.contrib.pytest。 img.png

但在实际使用过程中添加环境变量的方式好像没有生效。具体原因没有继续深入，可能是因为目前TFRobotServer使用的是pytest插件的方式来动态扩展环境变量，导致其加载时机不对（这是极大概率如此）。

最终推荐的方案是使用在conftest.py中添加pytest_plugins = ("celery.contrib.pytest", )的方式来启用celery.contrib.pytest。

如何对注册到特定app的任务进行完整的测试

参考官方文档，配合这个问题。可以解决 @shared_task 相关任务的处理。但是对于注册到特定app的任务，目前是通过自己阅读源码得到的一个解决方案，主要记录于此：Stackoverflow-西贝心合的回答。

其主要实现思路是参考celery_work这个fixture的实现，进行了自己的封装。这里面尚有如下问题未解决：

  1. celery.contrib.testing.tasks.ping 任务未自动注册的问题。因为ping任务没有自动注册，导致我们目前选择关掉了ping任务的测试。
  2. teardown时间的问题。默认teardown实现默认值是10秒，但在测试过程中发现，10秒有可能会引发一些异常，调整为30秒后，目前没有发现异常。至于触发异常的原因初步猜测可能是进程启动时间等等。但也有可疑之处，因为在目前测试的Mac Studio上，这台电脑的配置非常高，理论上启动一个进行不至于需要10秒的时间。但目前时间有限，未能完整解决这个问题，当前解决方案是加大Teardown的时间。