Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Mocking a Celery task call in Python with patch

Patching a Celery task call with a mocked return value returns <Mock name='mock().get()' ...> instead of the expected return_value defined by mock_task.get.return_value = "value". However, the mocked task functions correctly within my unit test.

Here is the unit test where I am patching my Celery task:

def test_foo(self):

    mock_task = Mock()
    mock_task.get = Mock(return_value={'success': True})

    print mock_task.get() # outputs {'success': True}

    with patch('app.tasks.my_task.delay', new=mock_task) as mocked_task:
        foo()  # this calls the mocked task with an argument, 'input from foo'
        mock_tasked.assert_called_with('input from foo')  # works

And here is the function being tested:

def foo():
    print tasks.my_task.delay  # shows a Mock object, as expected
    # now let's call get() on the mocked task:
    task_result = tasks.my_task.delay('input from foo').get()
    print task_result  # => <Mock name='mock().get()' id='122741648'>
    # unexpectedly, this does not return {'success': True}
    if task_result['success']:
        ...

The last line raises TypeError: 'Mock' object has no attribute '__getitem__'

Why can I call mock_task.get() from within my unit test, but calling it from foo returns a <Mock ...> instead of the expected return value?

like image 239
Petrus Theron Avatar asked Mar 27 '14 09:03

Petrus Theron


People also ask

How do you use a mock patch in Python?

How do we mock in Python? Mocking in Python is done by using patch to hijack an API function or object creation call. When patch intercepts a call, it returns a MagicMock object by default. By setting properties on the MagicMock object, you can mock the API call to return any value you want or raise an Exception .

What is Patch in mock?

patch() unittest. mock provides a powerful mechanism for mocking objects, called patch() , which looks up an object in a given module and replaces that object with a Mock . Usually, you use patch() as a decorator or a context manager to provide a scope in which you will mock the target object.

How does Celery execute tasks?

Process of Task Execution by Celery can be broken down into:Your application sends the tasks to the task broker, it is then reserved by a worker for execution & finally the result of task execution is stored in the result backend.


1 Answers

Unfortunately I know almost nothing about Celery, but it looks like the problem is with mocking.

You have:

tasks.my_task.delay('input from foo').get()

After patch('app.tasks.my_task.delay', new=mock_task) it becomes:

mock_task('input from foo').get()

Which is not the same as:

mock_task.get()

You should change your mock creation to:

mock_task().get = Mock(return_value={'success': True})

New Mock instance is created by default when you access existing Mock attribute or call it. So we can simplify it a little:

mock_task().get.return_value = {'success': True}
like image 197
and Avatar answered Oct 24 '22 17:10

and