在单元测试中修补时，芹菜任务调用的函数没有调用？答案

【问题标题】：Function called by a Celery task has no calls when patched in a unit test?在单元测试中修补时，芹菜任务调用的函数没有调用？
【发布时间】：2019-04-04 01:07:56
【问题描述】：

考虑以下tasks.py 模块（改编自http://docs.celeryproject.org/en/latest/getting-started/first-steps-with-celery.html#first-steps）：

import logging
import sys

from celery import Celery

app = Celery('tasks', broker='pyamqp://guest@localhost//')


@app.task
def add(x, y):
    logging.info(f"Adding {x} and {y}...")
    return x + y


def call_add(x, y):
    add.delay(x, y)

在同一个目录中，我有一个 test_tasks.py 测试模块，它读取

from unittest.mock import patch

import tasks


@patch('logging.info')
def test_adder(info_mock):
    tasks.call_add(1, 2)

    info_mock.assert_not_called()

此测试通过（如果我使用pytest test_tasks.py 运行它），但我不确定为什么没有调用info_mock？我希望以下断言能够通过

info_mock.assert_called_with("Adding 1 and 2...")

为什么在这个例子中没有通过tasks.call_add() 调用logging.info？在我看来，这相当于http://docs.celeryproject.org/en/latest/userguide/testing.html 中给出的示例。

【问题讨论】：

标签： python mocking celery python-unittest python-mock

【解决方案1】：

确保在运行单元测试时直接在同一进程中运行测试。

Celery 使得在“同步”运行任务时保持相同的 API 变得非常简单，并跳过损坏/工作部分。

app = Celery('tasks', broker='pyamqp://guest@localhost//', task_always_eager=True)

http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-always-eager

【讨论】：