セロリでのテスト

タスクと単体テスト

単体テストでタスクの動作をテストするには、モックを使用することをお勧めします。

熱心なモード

：setting： `task_always_eager` 設定によって有効になるイーガーモードは、定義上、単体テストには適していません。

イーガーモードでテストする場合、ワーカーで発生することのエミュレーションのみをテストすることになり、エミュレーションと実際に発生することの間には多くの矛盾があります。

熱心に実行されたタスクは、デフォルトではバックエンドに結果を書き込まないことに注意してください。この機能を有効にする場合は、：setting： `task_store_eager_result` をご覧ください。

Celeryタスクは、タスクとして呼び出されるコンテキストでアクションを実行する方法のみを定義する必要があるという点で、Webビューによく似ています。

これは、タスクがシリアル化、メッセージヘッダー、再試行などの処理のみを最適に処理し、実際のロジックが他の場所に実装されていることを意味します。

次のようなタスクがあったとします。

from .models import Product


@app.task(bind=True)
def send_order(self, product_pk, quantity, price):
    price = Decimal(price)  # json serializes this to string.

    # models are passed by id, not serialized.
    product = Product.objects.get(product_pk)

    try:
        product.order(quantity, price)
    except OperationalError as exc:
        raise self.retry(exc=exc)

Note：バインドされているのタスクは、タスクの最初の引数が常にタスクインスタンス（自己）になることを意味します。つまり、最初の引数として自己引数を取得し、タスククラスのメソッドと属性を使用できます。

この例のようにモックを使用して、このタスクの単体テストを作成できます。

from pytest import raises

from celery.exceptions import Retry

# for python 2: use mock.patch from `pip install mock`.
from unittest.mock import patch

from proj.models import Product
from proj.tasks import send_order

class test_send_order:

    @patch('proj.tasks.Product.order')  # < patching Product in module above
    def test_success(self, product_order):
        product = Product.objects.create(
            name='Foo',
        )
        send_order(product.pk, 3, Decimal(30.3))
        product_order.assert_called_with(3, Decimal(30.3))

    @patch('proj.tasks.Product.order')
    @patch('proj.tasks.send_order.retry')
    def test_failure(self, send_order_retry, product_order):
        product = Product.objects.create(
            name='Foo',
        )

        # Set a side effect on the patched methods
        # so that they raise the errors we want.
        send_order_retry.side_effect = Retry()
        product_order.side_effect = OperationalError()

        with raises(Retry):
            send_order(product.pk, 3, Decimal(30.6))

pytest

バージョン4.0の新機能。

Celeryは、統合（またはユニット）テストスイートで使用できるフィクスチャを追加する：pypi： `pytest` プラグインも利用できるようにします。

有効化

Celeryは最初、プラグインを無効な状態で出荷します。これを有効にするには、次のいずれかを実行できます。

pip install celery[pytest]

pip install pytest-celery

または環境変数を追加しますPYTEST_PLUGINS=celery.contrib.pytest

または、ルートconftest.pyにpytest_plugins = ("celery.contrib.pytest", )を追加します

マーク

celery-テストアプリの構成を設定します。

celeryマークを使用すると、単一のテストケースに使用される構成を上書きできます。

@pytest.mark.celery(result_backend='redis://')
def test_something():
    ...

または、クラス内のすべてのテストケースの場合：

@pytest.mark.celery(result_backend='redis://')
class test_something:

    def test_one(self):
        ...

    def test_two(self):
        ...

備品

関数スコープ

celery_app-テストに使用されるCeleryアプリ。

このフィクスチャは、テストに使用できるCeleryアプリを返します。

例：

def test_create_task(celery_app, celery_worker):
    @celery_app.task
    def mul(x, y):
        return x * y

    assert mul.delay(4, 4).get(timeout=10) == 16

celery_worker-ライブワーカーを埋め込みます。

このフィクスチャは、統合テストに使用できるCeleryワーカーインスタンスを開始します。ワーカーは別のスレッドで開始され、テストが戻るとすぐにシャットダウンされます。

デフォルトでは、フィクスチャはワーカーが未処理のタスクを完了するまで最大10秒待機し、制限時間を超えると例外を発生させます。タイムアウトは、celery_worker_parameters()フィクスチャによって返される辞書のshutdown_timeoutキーを設定することでカスタマイズできます。

例：

# Put this in your conftest.py
@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'redis://'
    }

def test_add(celery_worker):
    mytask.delay()


# If you wish to override some setting in one test cases
# only - you can use the ``celery`` mark:
@pytest.mark.celery(result_backend='rpc')
def test_other(celery_worker):
    ...

ハートビートはデフォルトで無効になっています。つまり、テストワーカーはworker-online、worker-offline、およびworker-heartbeatのイベントを送信しません。ハートビートを有効にするには、celery_worker_parameters()フィクスチャを変更します。

# Put this in your conftest.py
@pytest.fixture(scope="session")
def celery_worker_parameters():
    return {"without_heartbeat": False}
    ...

セッションスコープ

celery_config-オーバーライドしてCeleryテストアプリの構成をセットアップします。

このフィクスチャを再定義して、テストCeleryアプリを構成できます。

フィクスチャから返された設定は、celery_app()およびcelery_session_app()フィクスチャの設定に使用されます。

例：

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'rpc',
    }

celery_parameters-オーバーライドしてCeleryテストアプリのパラメーターを設定します。

このフィクスチャを再定義して、テストCeleryアプリの__init__パラメータを変更できます。 celery_config()とは対照的に、これらはCeleryをインスタンス化するときに直接渡されます。

フィクスチャから返された設定は、celery_app()およびcelery_session_app()フィクスチャの設定に使用されます。

例：

@pytest.fixture(scope='session')
def celery_parameters():
    return {
        'task_cls':  my.package.MyCustomTaskClass,
        'strict_typing': False,
    }

celery_worker_parameters-オーバーライドしてCeleryワーカーパラメーターを設定します。

このフィクスチャを再定義して、テストCeleryワーカーの__init__パラメーターを変更できます。これらは、インスタンス化されるときにWorkControllerに直接渡されます。

フィクスチャから返された設定は、celery_worker()およびcelery_session_worker()フィクスチャの設定に使用されます。

例：

@pytest.fixture(scope='session')
def celery_worker_parameters():
    return {
        'queues':  ('high-prio', 'low-prio'),
        'exclude_queues': ('celery'),
    }

celery_enable_logging-埋め込みワーカーへのログインを有効にするためにオーバーライドします。

これは、埋め込みワーカーへのログインを有効にするためにオーバーライドできるフィクスチャです。

例：

@pytest.fixture(scope='session')
def celery_enable_logging():
    return True

celery_includes-組み込みワーカーのインポートを追加します。

組み込みワーカーの起動時に、フィクスチャをオーバーライドしてモジュールを含めることができます。

これにより、インポートするモジュール名のリストを返すことができます。これには、タスクモジュール、シグナルを登録するモジュールなどがあります。

例：

@pytest.fixture(scope='session')
def celery_includes():
    return [
        'proj.tests.tasks',
        'proj.tests.celery_signal_handlers',
    ]

celery_worker_pool-組み込みワーカーに使用されるプールをオーバーライドします。

フィクスチャをオーバーライドして、組み込みワーカーに使用される実行プールを構成できます。

例：

@pytest.fixture(scope='session')
def celery_worker_pool():
    return 'prefork'

警告

gevent / eventletプールを使用することはできません。つまり、monkeypatchesを有効にしてテストスイート全体を実行している場合を除きます。

celery_session_worker-セッション全体に存在する組み込みワーカー。

このフィクスチャは、テストセッション全体を通して存続するワーカーを開始します（すべてのテストで開始/停止されるわけではありません）。

例：

# Add this to your conftest.py
@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'amqp://',
        'result_backend': 'rpc',
    }

# Do this in your tests.
def test_add_task(celery_session_worker):
    assert add.delay(2, 2) == 4

警告

セッションワーカーと一時的なワーカーを混在させるのはおそらく悪い考えです…

celery_session_app-テストに使用されるCeleryアプリ（セッションスコープ）。

これは、Celeryアプリインスタンスを参照する必要がある場合に、他のセッションスコープのフィクスチャで使用できます。

use_celery_app_trap-デフォルトアプリにフォールバックするときに例外を発生させます。

これは、conftest.pyでオーバーライドして、「アプリトラップ」を有効にすることができるフィクスチャです。何かがdefaultまたはcurrent_appにアクセスしようとすると、例外が発生します。

例：

@pytest.fixture(scope='session')
def use_celery_app_trap():
    return True

テストでデフォルトのアプリにアクセスする場合は、depends_on_current_appフィクスチャを使用してマークを付ける必要があります。

@pytest.mark.usefixtures('depends_on_current_app')
def test_something():
    something()

Celeryを使用したテスト—Pythonドキュメント

目次