Функция, вызываемая задачей Celery, не имеет вызовов при исправлении в модульном тесте? - PullRequest
Новая
       24

Функция, вызываемая задачей Celery, не имеет вызовов при исправлении в модульном тесте?

1 голос
/ 04 апреля 2019

Рассмотрим следующий 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.

1 Ответ

1 голос
/ 05 апреля 2019

Убедитесь, что тесты выполняются непосредственно в одном и том же процессе при выполнении юнит-тестов.

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

...