如何在单元测试期间捕获 Celery 任务？答案

【问题标题】：How do I capture Celery tasks during unit testing?如何在单元测试期间捕获 Celery 任务？
【发布时间】：2014-04-14 09:17:22
【问题描述】：

如何在不运行单元测试期间创建的 Celery 任务的情况下进行捕获？

例如，我想编写一个如下所示的测试：

def test_add_user_avatar():
    add_user_avatar(…)
    tasks = get_deferred_tasks(…)
    assert_equal(tasks[0], ResizeImageTask(…))

具体来说，我不想使用ALWAYS_EAGER——我的一些任务非常慢，并且有自己的一组测试用例。我特别想断言我的前端代码正在创建正确的任务。

【问题讨论】：

看看我的回答@stackoverflow.com/questions/4055860/…

标签： python celery

【解决方案1】：

我的情况类似，我正在使用的策略是模拟对 Celery 任务的调用，然后在运行后检查对这些模拟的调用。这可以在这里工作吗？

from … import ResizeImageTask


class NonQueuedTestCase(…):

    def setUp(self):
        """
        Patch out ResizeImageTask's delay method
        """
        super(NonQueuedTestCase, self).setUp()
        self.patcher = patch.object(ResizeImageTask, 'delay', autospec=True)
        self.m_delay = self.patcher.start()

    def tearDown(self):
        self.patcher.stop()
        super(NonQueuedTestCase, self).tearDown()

    def test_add_user_avatar(self):
        # Make call to front-end code
        add_user_avatar(…)
        # Check delay call was made
        self.m_delay.assert_called_once_with(…)

您可以在没有后端（在内存中或其他地方）的情况下运行这些测试，在前端代码和任务代码之间保持干净的分隔，并且可以测试多个代码路径，这些路径通常会在没有它的情况下排队一个长时间运行的任务正在运行。

【讨论】：