Блог

Рассмотрим следующий 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 позволяет очень просто сохранять одни и те же API во время выполнения задачи «синхронизированными» и пропускать нерабочую / рабочую часть.

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

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