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)

Answer 1

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 ===========================

Answer 2

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.