在 Celery 3.1.11 中使用 Django 进行单元测试？

Posted 2023-02-24

【中文标题】在 Celery 3.1.11 中使用 Django 进行单元测试？

【英文标题】：Unit Testing with Django in Celery 3.1.11?

【发布时间】：2014-08-04 07:03:10

【问题描述】：

我将 Celery 3.1.11 与 Django 1.6 一起使用。我一直在搜索，但我找不到很多关于芹菜的近期问题。(3.1)

我正在尝试通过 manage.py 运行单元测试：

>> python manage.py test calculationApp

在我的 testing.py for calculateApp 我创建了任务：

c = calculateCarbon.delay(project.id)
r = AsyncResult(c.id).ready()
print "c.backend: %s" % (c.backend)

print "AsyncResult(c.id).ready(): %s" % (r)
print "AsyncResult(c.id).state: %s" % (AsyncResult(c.id).state)
print "AsyncResult(c.id).result: %s" % (AsyncResult(c.id).result)   

while not r:
    r = AsyncResult(c.id).ready()

当我运行单元测试时，测试卡在测试中并且永远不会准备好（它永远不会超过 while 循环），我得到这个作为输出：

/usr/lib/python2.7/dist-packages/numpy/core/_methods.py:96: RuntimeWarning: invalid value encountered in double_scalars
ret = ret / float(rcount)

c.backend: None
AsyncResult(c.id).ready(): False
AsyncResult(c.id).state: PENDING
AsyncResult(c.id).result: None

此时我必须 CTRL+C 两次。

我正在阅读Celery 3.0 Docs - Unit Testing，它告诉我要设置。

CELERY_ALWAYS_EAGER = True
TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

这对 celery 3.1.11 仍然有效吗？我找不到有关 Django 单元测试的 Celery 3.1 的任何相关文档，而且我不确定这些设置是否有帮助或伤害，因为当我设置了这些设置后，任务的后端没有返回任何内容，但计算似乎实际执行.

当我从设置文件中删除这两行时，我会得到以下结果：

c.backend: <celery.backends.amqp.AMQPBackend object at 0x7a4de50>
AsyncResult(c.id).ready(): False
AsyncResult(c.id).state: PENDING
AsyncResult(c.id).result: None
AsyncResult(c.id).ready(): True
AsyncResult(c.id).state: FAILURE
AsyncResult(c.id).result: task args must be a list or tuple

======================================================================
FAIL: test_calculations (measuring.tests.TestCalculations)
----------------------------------------------------------------------
Traceback (most recent call last):
File "/var/www/project/calculationApp/tests.py", line 70, in test_calculations
self.assertEqual(int(number.attribute), 2212)
AssertionError: 0 != 2212

----------------------------------------------------------------------
Ran 1 test in 2.765s

【参考方案1】：

您是要测试实际的计算逻辑还是 Celery 有效的事实？这是两件不同的事情，在我看来，您应该非常努力地单独测试您的逻辑。它更快，更面向未来（如果您决定用其他东西代替 Celery 怎么办）并且大大简化了事情。 如果您坚持同时测试两者，只需在单元测试上下文中将 CELERY_ALWAYS_EAGER 设置为 True 即可解决您的问题。

您不需要测试 Celery，它可以工作，并且它有自己的测试套件来证明这一点。

【讨论】：

我只是想为我的应用程序运行单元测试，而不是测试 Celery 是否工作。我只是打印出任务的各种值，以了解 Celery 中发生了什么。