Mocking Celery `self.request` attribute for bound tasks when called directly

Disclaimer: Well, I do not think it was documented somewhere and this answer might be implementation-dependent.

Celery wraps his tasks into celery.Task instances, I do not know if it swaps the celery.Task.run method by the user task function or whatever.

But, when you call a task directly, you call __call__ and it'll push a context which will contain the task ID, etc…

So the idea is to bypass __call__ and Celery usual workings, first:

we push a controlled task ID: foobar.push_request(id=1) for example.
then, we call the run method: foobar.run(*args, **kwargs).

Example:

@app.task(bind=True)
def foobar(self, name):
    print(name)
    return foobar.utils.polling(self.request.id)

@patch('foobar.utils.polling')
def test_foobar(mock_polling):
    foobar.push_request(id=1)
    mock_polling.return_value = "done"
    assert foobar.run("test") == "done"
    mock_polling.assert_called_once_with(1)

Tags:

Unit Testing

Django

Celery

Related