Clever Geek Handbook

Erupting a call within a celery task

I have a flash application running a celery task. I'm trying to mock a single API call that goes deep inside this 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 would like to mock the string result = external_service()

so that I can run either the first or the second code.

So here is what I am 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 the Flask Python process, not the one Celery uses. Rejecting the task itself will not work as I am trying to test how the task is working when the external service returns 'x'

or 'y'

.

Help would be much appreciated.

+3


source to share


2 answers


A good option is to install CELERY_ALWAYS_EAGER

in True

your test configuration. It makes synchronous calls to celery. See the documentation for this option . With this option, any mockery created in your Flask process should work within the Celery task.

As a side benefit, your test configuration is simplified since you don't need to have a celery worker.

UPDATE : After discussion in the comments, apparently you don't want or cannot get rid of the Celery workers for testing configuration. In this case, I can offer three solutions, which I think do what you need:



  • Write a remote control command that mocks your celery task, that is, test code to run it on all workers with <a href = "http://celery.readthedocs.org/en/latest/userguide/workers.html#the-broadcast -function "rel =" nofollow "> broadcast ().

  • Define a custom command line parameter for your worker, say --test

    . Then add bootstep , which checks this argument and makes taunts.

  • Create an alternate module to give celery workers in a command line argument -A

    . It should be an identical copy of your original module, but with a mocking addition. Then run your workers with this alternate module for your tests.

I hope you find one of these three options satisfactory!

+5


source


Create setting 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
+5


source






More articles:


All Articles