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?
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 .
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.
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.
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}
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With