Celery 문서 는 Django 내에서 Celery 테스트를 언급 하지만 Django를 사용 하지 않는 경우 Celery 작업을 테스트하는 방법에 대해서는 설명하지 않습니다. 어떻게하나요?

답변

모든 unittest lib를 사용하여 동기식으로 작업을 테스트 할 수 있습니다. 나는 일반적으로 셀러리 작업을 할 때 두 가지 다른 테스트 세션을 수행합니다. 첫 번째는 (내가 제안한 것처럼) 완전히 동 기적이며 알고리즘이해야 할 일을 수행하는지 확인하는 것이어야합니다. 두 번째 세션은 전체 시스템 (브로커 포함)을 사용하고 직렬화 문제 나 기타 배포, 통신 문제가 없는지 확인합니다.

그래서:

from celery import Celery

celery = Celery()

@celery.task
def add(x, y):
    return x + y

그리고 테스트 :

from nose.tools import eq_

def test_add_task():
    rst = add.apply(args=(4, 4)).get()
    eq_(rst, 8)

도움이 되었기를 바랍니다.

답변

나는 이것을 사용한다 :

with mock.patch('celeryconfig.CELERY_ALWAYS_EAGER', True, create=True):
    ...

문서 : http://docs.celeryproject.org/en/3.1/configuration.html#celery-always-eager

CELERY_ALWAYS_EAGER를 사용하면 작업을 동기식으로 실행할 수 있으며 셀러리 서버가 필요하지 않습니다.

답변

정확히 무엇을 테스트하고 싶은지에 따라 다릅니다.

태스크 코드를 직접 테스트하십시오. “task.delay (…)”를 호출하지 말고 단위 테스트에서 “task (…)”를 호출하십시오.
CELERY_ALWAYS_EAGER를 사용하십시오 . 이렇게하면 “task.delay (…)”라고 말하는 지점에서 작업이 즉시 호출되므로 전체 경로를 테스트 할 수 있습니다 (비동기 동작은 아님).

답변

unittest

import unittest

from myproject.myapp import celeryapp

class TestMyCeleryWorker(unittest.TestCase):

  def setUp(self):
      celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)

py.test 픽스쳐

# conftest.py
from myproject.myapp import celeryapp

@pytest.fixture(scope='module')
def celery_app(request):
    celeryapp.conf.update(CELERY_ALWAYS_EAGER=True)
    return celeryapp

# test_tasks.py
def test_some_task(celery_app):
    ...

부록 : send_task를 열심히 존중하도록

from celery import current_app

def send_task(name, args=(), kwargs={}, **opts):
    # https://github.com/celery/celery/issues/581
    task = current_app.tasks[name]
    return task.apply(args, kwargs, **opts)

current_app.send_task = send_task

답변

Celery 4 사용자의 경우 :

@override_settings(CELERY_TASK_ALWAYS_EAGER=True)

설정 이름이 변경되었으며 업그레이드를 선택한 경우 업데이트가 필요하므로

https://docs.celeryproject.org/en/latest/history/whatsnew-4.0.html?highlight=what%20is%20new#lowercase-setting-names

답변

현재 셀러리 3.0 , 설정하는 하나의 방법 CELERY_ALWAYS_EAGER으로 장고 입니다 :

from django.test import TestCase, override_settings

from .foo import foo_celery_task

class MyTest(TestCase):

    @override_settings(CELERY_ALWAYS_EAGER=True)
    def test_foo(self):
        self.assertTrue(foo_celery_task.delay())

답변

Celery v4.0 이후 py.test 픽스쳐는 테스트를 위해 셀러리 워커를 시작하기 위해 제공 되며 완료되면 종료됩니다.

def test_myfunc_is_executed(celery_session_worker):
    # celery_session_worker: <Worker: gen93553@gnpill.local (running)>
    assert myfunc.delay().wait(3)

http://docs.celeryproject.org/en/latest/userguide/testing.html#py-test에 설명 된 다른 조명기 중에서 다음과 같이 조명기 를 재정 의하여 셀러리 기본 옵션을 변경할 수 있습니다 celery_config.

@pytest.fixture(scope='session')
def celery_config():
    return {
        'accept_content': ['json', 'pickle'],
        'result_serializer': 'pickle',
    }

기본적으로 테스트 작업자는 인 메모리 브로커와 결과 백엔드를 사용합니다. 특정 기능을 테스트하지 않는 경우 로컬 Redis 또는 RabbitMQ를 사용할 필요가 없습니다.