在 celery 任务中模拟调用答案

【问题标题】：Mocking out a call within a celery task在 celery 任务中模拟调用
【发布时间】：2015-05-26 05:53:15
【问题描述】：

我有一个运行 celery 任务的烧瓶应用程序。我正在尝试模拟在该任务深处发生的单个 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' 时任务的行为方式。

我们将不胜感激。

【问题讨论】：

标签： unit-testing flask celery python-mock

【解决方案1】：

一个不错的选择是在您的测试配置中将CELERY_ALWAYS_EAGER 设置为True。这使得对 Celery 的所有调用都是同步的。请参阅documentation for this option。使用此选项，您在 Flask 进程中设置的任何模拟都应该在 Celery 任务中工作。

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

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

编写一个模拟 Celery 任务的 remote control 命令，然后让测试代码在所有带有 broadcast() 的工作人员上运行它。
为您的工作人员定义一个custom command line option，例如--test。然后添加一个bootstep 来检查这个参数并进行模拟。
创建一个替代模块，以在-A 命令行参数中提供 Celery 工作人员。这应该是原始模块的相同副本，但添加了模拟。然后使用此替代模块启动您的工作人员进行测试。

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

【讨论】：

有没有办法在单次测试期间设置这个？我的套件中已经有测试异步功能的测试，因此在运行此测试时已经有一堆 celery worker 在运行。
如果你已经有Celery worker，那为什么不在worker进程中模拟这个任务呢？
好吧，我该怎么做？我不想在生产代码中添加任何东西，它必须在测试代码中完成。

【解决方案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

【讨论】：