如何正确地模拟在另一个celery任务中调用的celery任务?(下面是虚拟代码)
我有一个完全相同的代码结构在my_module.py中。位置在proj/cel/my_module.py。 我试着在proj/tests/cel_test/test.py里写测试。
测试函数:
@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)
我有一个完全相同的代码结构在my_module.py中。位置在proj/cel/my_module.py。 我试着在proj/tests/cel_test/test.py里写测试。
测试函数:
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)
delay
或apply
方法。 - Titouan Freville