unit-testing - 在 celery 任务中模拟调用

Question

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

View .py

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

任务模块.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 的所有调用都是同步的。见documentation for this option .使用此选项，您在 Flask 进程中设置的任何模拟都应该在 Celery 任务中工作。

附带的好处是，您的测试配置得到了简化，因为您不需要 Celery 工作人员。

更新 :经过评论中的讨论，您似乎不想或无法摆脱 Celery 工作人员的测试配置。在这种情况下，我可以提供三种我认为可以满足您需求的解决方案:

写一个remote control模拟您的 Celery 任务的命令，然后让测试代码使用 broadcast() 在所有工作人员上运行它.

定义一个 custom command line option对于您的员工，请说 --test .然后添加 bootstep检查此参数并进行模拟。

创建一个替代模块以提供 -A 中的 Celery worker 命令行参数。这应该是原始模块的相同副本，但添加了模拟。然后使用此替代模块启动您的工作人员进行测试。

我希望您能从这三个选项中找到满意的一个!