How Do I Capture Celery Tasks During Unit Testing?

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(): ad

Solution 1:

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


classNonQueuedTestCase(…):

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

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

    deftest_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.