在芹菜任务中嘲笑一个电话

Question

我有一个运行芹菜任务的烧瓶应用程序.我正在尝试模拟在该任务中发生的单个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过程中,而不是Celery正在使用的补丁.嘲讽任务本身并不像我想的测试工作是多么的任务时的行为外部服务回报'x'或'y'.

非常感谢帮助.

Answer 1

一个好的选择是设置CELERY_ALWAYS_EAGER到True您的测试配置.这使得对Celery的所有调用都是同步的.请参阅此选项的文档.使用此选项,您在Flask进程中设置的任何模拟都应该在Celery任务中运行.

作为附带好处,您的测试配置得以简化,因为您不需要Celery工作人员.

更新:在评论中讨论之后,您似乎不希望或无法摆脱Celery工作人员进行测试配置.在这种情况下,我可以提供三种解决方案,我认为可以满足您的需求:

1. 编写一个模拟Celery任务的远程控制命令,然后让测试代码在所有带有broadcast()的 worker上运行它.

2. 例如,为您的员工定义自定义命令行选项--test.然后添加一个检查此参数的bootstep并进行模拟.

3. 创建一个替代模块,以便在-A命令行参数中为Celery 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