84

我正在尝试为我们的django-celery项目提出一种测试方法。我已经阅读了文档中的注释，但它并没有让我对实际做什么有一个很好的了解。我不担心测试实际守护进程中的任务，只担心我的代码的功能。我主要想知道：

  1. 在测试期间我们如何绕过task.delay()（我尝试设置CELERY_ALWAYS_EAGER = True但没有任何区别）？
  2. 我们如何使用推荐的测试设置（如果这是最好的方法）而不实际更改我们的 settings.py？
  3. 我们还能使用manage.py test还是必须使用自定义跑步者？

总的来说，任何用芹菜测试的提示或技巧都会非常有帮助。

4

6 回答 6

73

我喜欢在需要完成 celery 结果的测试中使用 override_settings 装饰器。

from django.test import TestCase
from django.test.utils import override_settings
from myapp.tasks import mytask

class AddTestCase(TestCase):

    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
                       CELERY_ALWAYS_EAGER=True,
                       BROKER_BACKEND='memory')
    def test_mytask(self):
        result = mytask.delay()
        self.assertTrue(result.successful())

如果您想将此应用于所有测试，您可以使用http://docs.celeryproject.org/en/2.5/django/unit-testing.html中描述的 celery 测试运行器，它基本上设置了这些相同的设置，除了 ( BROKER_BACKEND = 'memory')。

在设置中：

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

查看 CeleryTestSuiteRunner 的源代码，很清楚发生了什么。

于 2012-12-19T05:08:11.037 回答
46

尝试设置：

BROKER_BACKEND = 'memory'

（感谢asksol的评论。）

于 2011-05-02T23:19:53.987 回答
18

这是我的测试基类的摘录，它将apply_async方法和记录到对它的调用（包括Task.delay.） 。这有点恶心，但在过去几个月我一直在使用它时，它设法满足了我的需求。

from django.test import TestCase
from celery.task.base import Task
# For recent versions, Task has been moved to celery.task.app:
# from celery.app.task import Task
# See http://docs.celeryproject.org/en/latest/reference/celery.app.task.html

class CeleryTestCaseBase(TestCase):

    def setUp(self):
        super(CeleryTestCaseBase, self).setUp()
        self.applied_tasks = []

        self.task_apply_async_orig = Task.apply_async

        @classmethod
        def new_apply_async(task_class, args=None, kwargs=None, **options):
            self.handle_apply_async(task_class, args, kwargs, **options)

        # monkey patch the regular apply_sync with our method
        Task.apply_async = new_apply_async

    def tearDown(self):
        super(CeleryTestCaseBase, self).tearDown()

        # Reset the monkey patch to the original method
        Task.apply_async = self.task_apply_async_orig

    def handle_apply_async(self, task_class, args=None, kwargs=None, **options):
        self.applied_tasks.append((task_class, tuple(args), kwargs))

    def assert_task_sent(self, task_class, *args, **kwargs):
        was_sent = any(task_class == task[0] and args == task[1] and kwargs == task[2]
                       for task in self.applied_tasks)
        self.assertTrue(was_sent, 'Task not called w/class %s and args %s' % (task_class, args))

    def assert_task_not_sent(self, task_class):
        was_sent = any(task_class == task[0] for task in self.applied_tasks)
        self.assertFalse(was_sent, 'Task was not expected to be called, but was.  Applied tasks: %s' %                 self.applied_tasks)

这是一个“头脑风暴”示例，说明如何在测试用例中使用它：

mymodule.py

from my_tasks import SomeTask

def run_some_task(should_run):
    if should_run:
        SomeTask.delay(1, some_kwarg=2)

test_mymodule.py

class RunSomeTaskTest(CeleryTestCaseBase):
    def test_should_run(self):
        run_some_task(should_run=True)
        self.assert_task_sent(SomeTask, 1, some_kwarg=2)

    def test_should_not_run(self):
        run_some_task(should_run=False)
        self.assert_task_not_sent(SomeTask)
于 2010-12-30T02:52:08.623 回答
4

因为我仍然在搜索结果中看到这个，设置覆盖

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner'

根据Celery Docs为我工作

于 2014-10-23T00:20:56.317 回答
2

对于 2019 年来到这里的每个人：查看这篇涵盖不同策略的文章，包括同步调用任务。

于 2019-04-17T04:25:28.907 回答
2

这就是我所做的

在 myapp.tasks.py 我有：

from celery import shared_task

@shared_task()
def add(a, b):
    return a + b

在 myapp.test_tasks.py 我有：

from django.test import TestCase, override_settings
from myapp.tasks import add


class TasksTestCase(TestCase):

    def setUp(self):
        ...

    @override_settings(CELERY_TASK_ALWAYS_EAGER=True,CELERY_TASK_EAGER_PROPOGATES=True)
    def test_create_sections(self):
        result= add.delay(1,2)
        assert result.successful() == True
        assert result.get() == 3
于 2019-06-04T13:33:15.457 回答