properly mock celery task that is being called inside another celery task

Question

How to properly mock celery task that is being called inside another celery task? (dummy code below)

@app.task
def task1(smthg):
    do_so_basic_stuff_1
    do_so_basic_stuff_2
    other_thing(smthg)

@app.task
def task2(smthg):
    if condition:
        task1.delay(smthg[1])
    else:
        task1.delay(smthg)

I do have exact same structure of code in my_module. proj/cel/my_module.py I'm trying to write test in proj/tests/cel_test/test.py

Test function:

def test_this_thing(self):
    # firs I want to mock task1
    # i've tried to import it from my_module.py to test.py and then mock it from test.py namespace 
    # i've tried to import it from my_module.py and mock it
    # nothing worked for me

    # what I basically want to do 
    # mock task1 here
    # and then run task 2 (synchronous)
    task2.apply()
    # and then I want to check if task one was called 
    self.assertTrue(mocked_task1.called)

kchomski · Accepted Answer · 2017-08-11 21:28:05Z

10

You are not calling task1() or task2(), but their methods: delay() and apply() - so you need to test if these methods get called.

Here is a working example I just wrote basing on your code:

tasks.py

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def task1():
    return 'task1'

@app.task
def task2():
    task1.delay()

test.py

from tasks import task2

def test_task2(mocker):
    mocked_task1 = mocker.patch('tasks.task1')
    task2.apply()
    assert mocked_task1.delay.called

Test results:

$ pytest -vvv test.py
============================= test session starts ==============================
platform linux -- Python 3.5.2, pytest-3.2.1, py-1.4.34, pluggy-0.4.0 -- /home/kris/.virtualenvs/3/bin/python3
cachedir: .cache
rootdir: /home/kris/projects/tmp, inifile:
plugins: mock-1.6.2, celery-4.1.0
collected 1 item                                                                

test.py::test_task2 PASSED

=========================== 1 passed in 0.02 seconds ===========================

answered Aug 11, 2017 at 21:28

kchomski

3,07023 silver badges34 bronze badges

Sign up to request clarification or add additional context in comments.

1 Comment

Titouan Freville Over a year ago

For a more complete testing, you can also mock directly the delay or apply method.

MrName · Accepted Answer · 2017-08-11 14:32:39Z

4

To start, testing Celery tasks can be REALLY difficult. I generally put all of my logic into a function that is NOT a task, and then make a task that just calls that function, so that you can properly test the logic.

Second, I don't think you want to be calling tasks inside of tasks (not certain, but I believe this is generally not recommended). Instead, depending on your needs, you should probably be chaining or grouping:

http://docs.celeryproject.org/en/latest/userguide/canvas.html#the-primitives

Lastly, to answer your actual question, you would want to patch the delay method exactly where it occurs in your code, as described in this post.

answered Aug 11, 2017 at 14:32

MrName

2,53918 silver badges32 bronze badges

1 Comment

zerohedge Over a year ago

Nothing wrong with calling tasks inside of other tasks as long as you're not waiting for it (using its result).

Collectives™ on Stack Overflow

properly mock celery task that is being called inside another celery task

2 Answers 2

1 Comment

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

1 Comment

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related