다음 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
이 기사는 인터넷에서 수집됩니다. 재 인쇄 할 때 출처를 알려주십시오.
침해가 발생한 경우 연락 주시기 바랍니다[email protected] 삭제
몇 마디 만하겠습니다