如何在单元测试期间捕获Celery任务? [英] How do I capture Celery tasks during unit testing?

问题描述

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

How can I capture without running Celery tasks created during a unit test?

例如，我想编写一个看起来像这样的测试:

For example, I'd like to write a test which looks something like this:

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

具体来说，我不想不想使用 ALWAYS_EAGER -我的某些任务很慢，并且有自己的测试用例集.我特别想断言我的前端代码正在创建正确的任务.

Specifically, I do not want to use ALWAYS_EAGER — some of my tasks are quite slow, and have their own set of tests cases. I specifically want to assert that the correct tasks are being created by my front-end code.

推荐答案

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

My situation is similar and the strategy I'm working with is to mock out the calls to Celery tasks and then check the calls made to those mocks after the run. Could this work here?

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(…)

您可以在没有后端(内存或其他方式)的情况下运行这些测试，在前端代码和任务代码之间保持清晰的间隔，并可以测试通常会使长时间运行的任务排队而没有它的多个代码路径运行.

You can run these tests without a backend up (in memory or otherwise), keep a clean break between the front-end code and task code, and can test multiple code paths that would normally queue up a long running task without it running.

这篇关于如何在单元测试期间捕获Celery任务?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！