使用 Celery 进行测试¶

Testing with Celery

中文

使用 Celery 进行测试分为两部分：

单元测试与集成测试：使用 celery.contrib.pytest。
烟雾测试 / 生产环境测试：使用 pytest-celery >= 1.0.0

安装 pytest-celery 插件时，还会同时安装 celery.contrib.pytest 基础设施，以及 pytest 插件基础设施。它们的区别在于使用方式。

警告

两个 API 之间是不兼容的。pytest-celery 插件基于 Docker，而 celery.contrib.pytest 则基于 mock。

要使用 celery.contrib.pytest 基础设施，请按照以下说明进行操作。

pytest-celery 插件有其独立文档。

英文

Testing with Celery is divided into two parts:

Unit & Integration: Using celery.contrib.pytest.
Smoke / Production: Using pytest-celery >= 1.0.0

Installing the pytest-celery plugin will install the celery.contrib.pytest infrastructure as well, alongside the pytest plugin infrastructure. The difference is how you use it.

警告

Both APIs are NOT compatible with each other. The pytest-celery plugin is Docker based and the celery.contrib.pytest is mock based.

To use the celery.contrib.pytest infrastructure, follow the instructions below.

The pytest-celery plugin has its own documentation.

任务和单元测试¶

Tasks and unit tests

中文

在单元测试中测试任务行为时，推荐的方法是使用 mock。

急切模式

task_always_eager 设置启用的急切模式定义上并不适合单元测试。

使用急切模式测试时，实际上你仅仅是在测试一个仿真环境与工作进程之间的差异，而仿真与实际工作中所发生的行为之间存在许多不一致。

注意，急切执行的任务默认不会将结果写入后端。如果你想启用此功能，可以查看 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 序列化为字符串。

    # 模型通过 ID 传递，而非序列化。
    product = Product.objects.get(product_pk)

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

Note: 任务被绑定意味着任务的第一个参数将始终是任务实例（self）。这意味着你会得到一个 self 参数作为第一个参数，可以使用 Task 类的方法和属性。

你可以为此任务编写单元测试，使用像下面这样的 mock：

from pytest import raises

from celery.exceptions import Retry

# 对于 Python 2：使用 mock.patch 从 `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')  # < 在上面模块中对 Product 进行 patch
    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',
        )

        # 设置 patch 方法的副作用
        # 使它们抛出我们需要的错误。
        send_order_retry.side_effect = Retry()
        product_order.side_effect = OperationalError()

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

英文

To test task behavior in unit tests the preferred method is mocking.

Eager mode

The eager mode enabled by the task_always_eager setting is by definition not suitable for unit tests.

When testing with eager mode you are only testing an emulation of what happens in a worker, and there are many discrepancies between the emulation and what happens in reality.

Note that eagerly executed tasks don't write results to backend by default. If you want to enable this functionality, have a look at task_store_eager_result.

A Celery task is much like a web view, in that it should only define how to perform the action in the context of being called as a task.

This means optimally tasks only handle things like serialization, message headers, retries, and so on, with the actual logic implemented elsewhere.

Say we had a task like this:

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: A task being bound means the first argument to the task will always be the task instance (self). which means you do get a self argument as the first argument and can use the Task class methods and attributes.

You could write unit tests for this task, using mocking like in this example:

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))

使用 Celery 进行测试¶

任务和单元测试¶

pytest¶

启用¶

标记¶

`celery` - 设置测试应用配置。¶

Fixtures¶

函数作用域¶

`celery_app` - 用于测试的 Celery 应用。¶

`celery_worker` - 嵌入实时工作线程。¶

会话作用域¶

`celery_config` - 覆盖该函数以设置 Celery 测试应用配置。¶

`celery_parameters` - 覆盖该函数以设置 Celery 测试应用参数。¶

`celery_worker_parameters` - 覆盖该函数以设置 Celery 工作线程参数。¶

`celery_enable_logging` - 覆盖该函数以启用嵌入式工作线程的日志记录。¶

`celery_includes` - 为嵌入式工作线程添加额外的导入。¶

`celery_worker_pool` - 覆盖用于嵌入式工作线程的池。¶

`celery_session_worker` - 在整个会话期间都处于活动状态的嵌入式工作线程。¶

`celery_session_app` - 用于测试的 Celery 应用（会话范围）。¶

`use_celery_app_trap` - 回退到默认应用时引发异常。¶

使用 Celery 进行测试¶

任务和单元测试¶

pytest¶

启用¶

标记¶

celery - 设置测试应用配置。¶

Fixtures¶

函数作用域¶

celery_app - 用于测试的 Celery 应用。¶

celery_worker - 嵌入实时工作线程。¶

会话作用域¶

celery_config - 覆盖该函数以设置 Celery 测试应用配置。¶

celery_parameters - 覆盖该函数以设置 Celery 测试应用参数。¶

celery_worker_parameters - 覆盖该函数以设置 Celery 工作线程参数。¶

celery_enable_logging - 覆盖该函数以启用嵌入式工作线程的日志记录。¶

celery_includes - 为嵌入式工作线程添加额外的导入。¶

celery_worker_pool - 覆盖用于嵌入式工作线程的池。¶

celery_session_worker - 在整个会话期间都处于活动状态的嵌入式工作线程。¶

celery_session_app - 用于测试的 Celery 应用（会话范围）。¶

use_celery_app_trap - 回退到默认应用时引发异常。¶

`celery` - 设置测试应用配置。¶

`celery_app` - 用于测试的 Celery 应用。¶

`celery_worker` - 嵌入实时工作线程。¶

`celery_config` - 覆盖该函数以设置 Celery 测试应用配置。¶

`celery_parameters` - 覆盖该函数以设置 Celery 测试应用参数。¶

`celery_worker_parameters` - 覆盖该函数以设置 Celery 工作线程参数。¶

`celery_enable_logging` - 覆盖该函数以启用嵌入式工作线程的日志记录。¶

`celery_includes` - 为嵌入式工作线程添加额外的导入。¶

`celery_worker_pool` - 覆盖用于嵌入式工作线程的池。¶

`celery_session_worker` - 在整个会话期间都处于活动状态的嵌入式工作线程。¶

`celery_session_app` - 用于测试的 Celery 应用（会话范围）。¶

`use_celery_app_trap` - 回退到默认应用时引发异常。¶