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

Viewed 5563

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)
2 Answers
Related