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

Question

I have a task foobar:

@app.task(bind=True)
def foobar(self, owner, a, b):
   if already_working(owner): # check if a foobar task is already running for owner.
       register_myself(self.request.id, owner) # add myself in the DB.
   return a + b

How can I mock the self.request.id attribute? I am already patching everything and calling directly the task rather than using .delay/.apply_async, but the value of self.request.id seems to be None (as I am doing real interactions with DB, it is making the test fail, etc…).

For the reference, I'm using Django as a framework, but I think that this problem is just the same, no matter the environment you're using.

This doesn't make sense. Is that task part of a class? If so, show it; if not, it wouldn't have a `self`. And in any case, a task does not run as part of a request, so why would there be a `self.request`? — Daniel Roseman, Oct 29 '17 at 13:13
@DanielRoseman Did you already use Celery? If not, please read this beforehand: http://docs.celeryproject.org/en/latest/userguide/tasks.html#bound-tasks ; the decorator is wrapping the function in a class. — Raito, Oct 29 '17 at 13:16

score 5 · Accepted Answer · answered Oct 29 '17 at 13:58

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)

score 2 · Answer 2 · answered Oct 17 '19 at 09:30

2

You can call the task synchronously using

task = foobar.s(<args>).apply()

This will assign a unique task ID, so the value will not be None and your code will run. Then you can check the results as part of your test.

answered Oct 17 '19 at 09:30

ronathan

736
7
5

Stuart Buckingham · Answer 3 · 2021-01-25T20:25:29.050

0

There is probably a way to do this with patch, but I could not work out a way to assign a property. The most straightforward way is to just mock self.

tasks.py:

@app.task(name='my_task')
def my_task(self, *args, **kwargs):
    *__do some thing__*

test_tasks.py:

from mock import Mock

def test_my_task():
    self = Mock()
    self.request.id = 'ci_test'
    my_task(self)

edited Jan 25 '21 at 20:25

answered Jan 25 '21 at 02:57

Stuart Buckingham

1,574
16
25

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

3 Answers3

Linked