单元测试——模拟芹菜任务中的调用

Question

我有一个运行芹菜任务的flask应用程序。我试着模拟一个发生在该任务深处的单个API调用。

views.py

from mypackage.task_module import my_task
@app.route('/run_task')
def run_task():
    task = my_task.delay()
    return some_response

task_module.py

from mypackage.some_module import SomeClass
@celery.task
def my_task():
    return SomeClass().some_function()

some_module.py

from mypackage.xyz import external_service
class SomeClass(object):
    def some_function(self):
        #do some stuff
        result = external_service(some_param)
        if 'x' in result:
             #do something
        elif 'y' in result:
             #do something else

我想模拟result = external_service()行，这样我就可以触发第一个或第二个代码路径。

这就是我要做的:

@mock.patch('mypackage.some_module.external_service', autospec=True)
def test_x_path(my_mock):
    my_mock.return_value = {'x': some_val}
    #run test, expect 'x' code path to run

然而，这不起作用，因为(我认为)补丁发生在Flask的Python进程中，而不是芹菜使用的那个。mock任务本身不会工作，因为我要测试的是当外部服务返回'x'或'y'时任务的行为。

Answer 1

一个好的选择是在测试配置中将CELERY_ALWAYS_EAGER设置为True。这使得所有对芹菜的调用都是同步的。请参阅该选项的文档。有了这个选项，你在Flask进程中设置的任何mock都应该在一个芹菜任务中工作。

作为附带的好处，您的测试配置被简化了，因为您不需要有一个Celery worker。

UPDATE:在评论中的讨论之后，似乎你不想，或者不能为你的测试配置摆脱芹菜工作器。在这种情况下，我可以提供三种解决方案，我认为它们可以满足您的需要:

1. 写一个远程控制命令来模拟你的芹菜任务，然后让测试代码用broadcast()在你所有的工作上运行它。

2. 为您的worker定义一个自定义命令行选项，例如--test。然后添加一个引导步骤来检查这个参数并进行模拟。

3. 创建一个替代模块，在-A命令行参数中提供芹菜工作器。这应该是原始模块的相同副本，但添加了mock。然后使用此可选模块启动您的worker，用于测试。

我希望你在这三个选项中找到一个满意的!

Answer 2

创建测试函数

class TestCeleryTask(TestCase):
    def setUp(self):
         app.config['CELERY_ALWAYS_EAGER'] = True
         app.config['BROKER_BACKEND'] = 'memory'
         app.config['CELERY_EAGER_PROPAGATES_EXCEPTIONS'] = True
    def test_task(self):
         # test it