lundi 25 mai 2015

Mocking out a call within a celery task

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')
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.

Aucun commentaire:

Enregistrer un commentaire