Mocking out a call within a celery task

Question

I have a flask app that runs a celery task. I'm trying to mock out a single API call that happens deep within that task.

views.py

from mypackage.task_module import my_task
@app.route('/run_task')
def run_task():
    task = my_task.delay()
    return some_response

task_module.py

from mypackage.some_module import SomeClass

@celery.task
def my_task():
    return SomeClass().some_function()

some_module.py

from mypackage.xyz import external_service
class SomeClass(object):
    def some_function(self):
        #do some stuff
        result = external_service(some_param)
        if 'x' in result:
             #do something
        elif 'y' in result:
             #do something else

I'd like to mock out the result = external_service() line so I can trigger either the first or the second code path.

So here's what I'm trying:

@mock.patch('mypackage.some_module.external_service', autospec=True)
def test_x_path(my_mock):
    my_mock.return_value = {'x': some_val}
    #run test, expect 'x' code path to run

However, this doesn't work, because (I think) the patch happens in Flask's Python process, and not the one that Celery is using. Mocking the task itself won't work as what I'm trying to test is how the task behaves when the external service returns 'x' or 'y'.

Help would be much appreciated.

Answer 1

A good option is to set CELERY_ALWAYS_EAGER to True in your test configuration. This makes all calls to Celery synchronous. See the documentation for this option. With this option any mocking you set up in your Flask process should work within a Celery task.

As a side benefit, your testing configuration is simplified, as you don't need to have a Celery worker.

UPDATE: After the discussion in the comments, it appears you do not want to, or can't get rid of the Celery workers for your testing configuration. In that case I can offer three solutions that I think do what you need:

1. Write a remote control command that mocks your Celery task, then have the test code run it on all your workers with broadcast().

2. Define a custom command line option for your worker, say --test. Then add a bootstep that checks for this argument and does the mocking.

3. Create an alternative module to give the Celery workers in the -A command line argument. This should be an identical copy of your original module, but with the mocking added. Then start your workers with this alternative module for your tests.

I hope you find one of these three options satisfactory!

Answer 2

Create a set up for test function

class TestCeleryTask(TestCase):
    def setUp(self):
         app.config['CELERY_ALWAYS_EAGER'] = True
         app.config['BROKER_BACKEND'] = 'memory'
         app.config['CELERY_EAGER_PROPAGATES_EXCEPTIONS'] = True

    def test_task(self):
         # test it