在单元测试中打补丁时,Celery 任务调用的函数没有调用?
Function called by a Celery task has no calls when patched in a unit test?
考虑以下 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...")
为什么在这个例子中logging.info
不是通过tasks.call_add()
调用的?在我看来相当于http://docs.celeryproject.org/en/latest/userguide/testing.html.
中给出的例子
确保 运行 在 运行 单元测试时直接在同一进程中进行测试。
Celery 使得在 运行 执行任务 "in-sync" 并跳过 broken/worker 部分时保持相同的 API 变得非常简单。
app = Celery('tasks', broker='pyamqp://guest@localhost//', task_always_eager=True)
http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-always-eager
考虑以下 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...")
为什么在这个例子中logging.info
不是通过tasks.call_add()
调用的?在我看来相当于http://docs.celeryproject.org/en/latest/userguide/testing.html.
确保 运行 在 运行 单元测试时直接在同一进程中进行测试。
Celery 使得在 运行 执行任务 "in-sync" 并跳过 broken/worker 部分时保持相同的 API 变得非常简单。
app = Celery('tasks', broker='pyamqp://guest@localhost//', task_always_eager=True)
http://docs.celeryproject.org/en/latest/userguide/configuration.html#task-always-eager