Я хочу проверить, что конкретная задача сельдерея вызывается 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()
.
Заранее спасибо за любой ответ.