How to mock the bound context of a celery task

Question

How do I mock the bound context, or mock the celery task id?

Given a celery task like:

helpers.py:

from task import some_task

def some_helper():
    some_task.delay(123)

in task.py:

@app.task(queue="abc", bind=True)
def some_task(self, some_number: int):
    print(self.id) # how to mock this attribute access?

Simple test case:

from django.test.testcases import TestCase
from helpers import some_helper


class SomeTest(TestCase):

    def test_some_helper(self):
        some_helper()

I tried:

 @patch("celery.app.base.Celery.task", return_value=lambda x: x)

I also tried:

class MockResult(dict):
    def __getattr__(self, x):
        return self[x]

...
def test_some_task(self):
    cls = MockResult({"id": "asdf"})
    bound_some_task = some_task.__get__(cls, MockResult)
    bound_some_task(123)

Garry Polley · Accepted Answer · 2020-01-23 01:27:02Z

Given a celery task that looks like:

@my_celery_app.task(bind=True)
def my_task(self):
    if self.request.retries == 1:
        my_method_to_invoke()
        # Do work for first retry
    elif self.request.retries == 2:
        # Do work for second retry
    # do work for main task

The test can set the self.request.retries by mocking the base Task.request class attribute within celery.

In the unit test the following can be done

@patch("path.to.task.my_method_to_invoke")
@patch("celery.app.task.Task.request")
def my_test_method(self, mock_task_request, mock_my_method_to_invoke):

    # Set the value of retries directly
    mock_task_request.retries = 1

    # Call the task and assert the inside method was
    # called
    my_task()

    mock_my_method_to_invoke.assert_called_once()

It may be possible to do the same with id on Task. I was lead to this answer looking for how to mock the self on a bound celery task.

jmunsch · Accepted Answer · 2019-04-24 16:35:24Z

1

Was able to get something working by using setattr on the task method, not sure if there is a better/other ways to do this:

from django.test.testcases import TestCase
from helpers import some_helper

class SomeTest(TestCase):

    def test_some_helper(self):

        from task import some_task
        setattr(some_task, 'id', 'hello-id')

        some_helper()

In addition to this it is possible to mock the request.id or "task id" like so:

@patch("task.some_task.request.id", return_value="hello-id")
def test_some_helper(...): ....

edited Apr 24, 2019 at 16:35

answered Apr 23, 2019 at 22:46

jmunsch

24.3k12 gold badges102 silver badges120 bronze badges

Comments

Just-Drue · Accepted Answer · 2020-01-29 11:30:11Z

1

What helped me was to create a mock class mimicking the task

class CeleryTaskHelper:
    """Mock task id"""

    class Request:
        @property
        def id(self):
            return ''.join(random.choices(string.ascii_uppercase + string.digits, k=20))

    @property
    def request(self):
        return self.Request()

    def revoke(self):
        return

and then

@patch('apps.orders.tasks.activate_order.apply_async', return_value=CeleryTaskHelper())

answered Jan 29, 2020 at 11:30

Just-Drue

535 bronze badges

1 Comment

Ron Over a year ago

i didnt even need the new class, just adding .apply_async to the patch string does the magic. thx!

ronathan · Accepted Answer · 2019-10-17 09:56:26Z

0

I tried setattr and the patch annotation, but neither worked for me... seemingly because "request.id" is a nested object, and I couldn't get that working with setattr (though there is supposed to be a way).

One options is to call the task synchronously using

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

This will assign a unique task ID.

Another option is to have the task be a wrapper method, such as:

@app.task
def some_task(self, ...args):
    some_method(self.request.id, ...args)

And then you can test some_method directly, passing in whatever ID you like.

answered Oct 17, 2019 at 9:56

ronathan

7467 silver badges5 bronze badges

Collectives™ on Stack Overflow

How to mock the bound context of a celery task

4 Answers 4

Comments

Comments

1 Comment

Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

4 Answers 4

Comments

Comments

1 Comment

Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related