How to fix Celery tasks during unit testing?

Question

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

For example, I would like to write a test that looks something like this:

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

In particular, I do not want to use it ALWAYS_EAGER- some of my tasks are rather slow and have their own set of tests. I specifically want to argue that the right tasks are created by my interface.

Answer 1

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

from … import ResizeImageTask


class NonQueuedTestCase(…):

    def setUp(self):
        """
        Patch out ResizeImageTask 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(…)

( ), , , .