34

我有一个用 Django 编写的 REST API，其端点在发布到芹菜任务时对其进行排队。响应包含我想用来测试任务是否已创建并获得结果的任务 ID。所以，我想做类似的事情：

def test_async_job():
    response = self.client.post("/api/jobs/", some_test_data, format="json")
    task_id = response.data['task_id']
    result = my_task.AsyncResult(task_id).get()
    self.assertEquals(result, ...)

我显然不想让 celery worker 运行单元测试，我希望以某种方式模拟它。我不能使用CELERY_ALWAYS_EAGER因为这似乎完全绕过了代理，阻止我使用 AsyncResult 通过其 ID 获取任务（如此处所述）。

通过 celery 和kombu docs，我发现有一个用于单元测试的内存传输，可以满足我的需求。我尝试覆盖BROKER_URL设置以在测试中使用它：

@override_settings(BROKER_URL='memory://')
def test_async_job():

但行为与 ampq 代理相同：它阻止测试等待结果。任何想法我应该如何配置这个代理让它在测试中工作？

4

3 回答 3

19

您可以在设置中指定 broker_backend ：

if 'test' in sys.argv[1:]:
    BROKER_BACKEND = 'memory'
    CELERY_TASK_ALWAYS_EAGER = True
    CELERY_TASK_EAGER_PROPAGATES = True

或者您可以直接在测试中使用装饰器覆盖设置

import unittest
from django.test.utils import override_settings


class MyTestCase(unittest.TestCase):

    @override_settings(CELERY_TASK_EAGER_PROPAGATES=True,
                       CELERY_TASK_ALWAYS_EAGER=True,
                       BROKER_BACKEND='memory')
    def test_mytask(self):
        ...
于 2014-10-03T06:31:20.780 回答
13

您可以使用 Kombu 内存代理来运行单元测试，但是要这样做，您需要使用与 Django 服务器相同的 Celery 应用程序对象来启动一个 Celery 工作程序。

要使用内存代理，请将 BROKER_URL 设置为memory://localhost/

然后，要启动一个小型芹菜工人，您可以执行以下操作：

app = <Django Celery App>

# Set the worker up to run in-place instead of using a pool
app.conf.CELERYD_CONCURRENCY = 1
app.conf.CELERYD_POOL = 'solo'

# Code to start the worker
def run_worker():
    app.worker_main()

# Create a thread and run the worker in it
import threading
t = threading.Thread(target=run_worker)
t.setDaemon(True)
t.start()

您需要确保使用与 Django celery 应用程序实例相同的应用程序。

请注意，启动工作程序将打印许多内容并修改日志记录设置。

于 2015-08-20T10:10:02.160 回答
11

TransactionTestCase这是一个与 Celery 4.x 一起工作的 Django 的更全面的示例。

import threading

from django.test import TransactionTestCase
from django.db import connections

from myproj.celery import app  # your Celery app


class CeleryTestCase(TransactionTestCase):
    """Test case with Celery support."""

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        app.control.purge()
        cls._worker = app.Worker(app=app, pool='solo', concurrency=1)
        connections.close_all()
        cls._thread = threading.Thread(target=cls._worker.start)
        cls._thread.daemon = True
        cls._thread.start()

    @classmethod
    def tearDownClass(cls):
        cls._worker.stop()
        super().tearDownClass()

请注意，这不会将您的队列名称更改为测试队列，因此如果您也在运行该应用程序，您也需要这样做。

于 2017-04-27T04:55:04.633 回答