Невозможно исправить logging.Logger.info вызвано внутри задачи сельдерея - PullRequest
0 голосов
/ 04 мая 2019

Я хочу проверить, что конкретная задача сельдерея вызывается logger.info ровно один раз, когда задача вызывается с помощью delay() API.

И я хочу выполнить тест путем установки патча logger.info.

Я хочу протестировать, как описано здесь, для случая Product.order https://docs.celeryproject.org/en/latest/userguide/testing.html.

Установка: python 2.7 в Ubuntu 16.04.celery 4.3.0, pytest 4.0.0, mock 3.0.3.

У меня следующая структура файловой системы:

poc/
   prj/
     -celery_app.py
     -tests.py

celery_app.py

from __future__ import absolute_import

from celery import Celery

from celery.utils.log import get_task_logger

app = Celery('celery_app')

logger = get_task_logger(__name__)


@app.task(bind=True)
def debug_task(self):

    logger.info('Request: {0!r}'.format(self.request))

tests.py

from __future__ import absolute_import

from mock import patch

from prj.celery_app import debug_task


@patch('logging.Logger.info')
def test_log_info_is_called_only_once_when_called_sync(log_info):
    debug_task()
    log_info.assert_called_once()


@patch('logging.Logger.info')
def test_log_info_is_called_only_once_when_called_async(log_info):
    debug_task.delay()
    log_info.assert_called_once()

Я ожидаю, что оба теста пройдут успешно.Вместо этого первое имеет успех, а второе - с AssertionError: Expected 'info' to have been called once. Called 0 times.

Я ожидаю, что оценка выражения logger.info внутри debug_task() context в обоих случаях оценивается как <MagicMock name='info' id='someinteger'>, вместо этого <bound method Logger.info of <logging.Logger object at 0x7f894eeb1690>> во втором случае, не показывая исправлений.

Я знаю, что во втором случае работник сельдерея выполняет задачу внутри потока.

Я прошу способ пропатчить вызов logger.info при выполнении debug_task.delay().

Заранее спасибо за любой ответ.

...