使 Django 测试用例数据库对 Celery 可见

Posted 2023-02-24

【中文标题】使 Django 测试用例数据库对 Celery 可见

【英文标题】：Make Django test case database visible to Celery

【发布时间】：2018-03-13 20:03:26

【问题描述】：

当 Django 测试用例运行时，它会创建一个隔离的测试数据库，以便在每个测试完成时回滚数据库写入。我正在尝试使用 Celery 创建一个集成测试，但我不知道如何将 Celery 连接到这个临时测试数据库。在幼稚的设置中，保存在 Django 中的对象对 Celery 是不可见的，而保存在 Celery 中的对象会无限期地存在。

这是一个示例测试用例：

import json
from rest_framework.test import APITestCase
from myapp.models import MyModel
from myapp.util import get_result_from_response

class MyTestCase(APITestCase):
    @classmethod
    def setUpTestData(cls):
        # This object is not visible to Celery
        MyModel(id='test_object').save()

    def test_celery_integration(self):
        # This view spawns a Celery task
        # Task should see MyModel.objects.get(id='test_object'), but can't
        http_response = self.client.post('/', 'test_data', format='json')

        result = get_result_from_response(http_response)
        result.get()  # Wait for task to finish before ending test case
        # Objects saved by Celery task should be deleted, but persist

我有两个问题：

如何使 Celery 可以看到 Django 测试用例的对象？

如何确保 Celery 保存的所有对象在测试完成后自动回滚？

如果无法自动执行此操作，我愿意手动清理对象，但删除 tearDown 中的对象，即使是在 APISimpleTestCase 中似乎也会回滚。

【参考方案1】：

这可以通过在 Django 测试用例中启动一个 Celery worker 来实现。

背景

Django 的内存数据库是 sqlite3。正如它在the description page for Sqlite in-memory databases 上所说，“[A]ll 共享内存数据库的所有数据库连接都需要在同一个进程中。”这意味着，只要 Django 使用内存中的测试数据库，并且 Celery 是在单独的进程中启动的，那么 Celery 和 Django 根本不可能共享一个测试数据库。

但是，使用celery.contrib.testing.worker.start_worker，可以在同一进程内的单独线程中启动 Celery 工作程序。该工作人员可以访问内存数据库。

这假设 Celery 已经在 the usual way 中与 Django 项目一起设置。

解决方案

因为 Django-Celery 涉及到一些跨线程通信，所以只有不在隔离事务中运行的测试用例才会起作用。测试用例必须直接从SimpleTestCase 或其等效的APISimpleTestCase 继承，并将databases 设置为'__all__' 或只是与测试交互的数据库。

关键是在TestCase的setUpClass方法中启动一个Celery worker，在tearDownClass方法中关闭它。关键函数是celery.contrib.testing.worker.start_worker，需要当前Celery应用的一个实例，大概是从mysite.celery.app获取的，返回一个PythonContextManager，里面有__enter__和__exit__方法，必须在@987654336中调用@ 和 tearDownClass，分别。可能有一种方法可以避免使用装饰器或其他东西手动输入和存在ContextManager，但我无法弄清楚。这是一个示例tests.py 文件：

from celery.contrib.testing.worker import start_worker
from django.test import SimpleTestCase

from mysite.celery import app

class BatchSimulationTestCase(SimpleTestCase):
    databases = '__all__'

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        # Start up celery worker
        cls.celery_worker = start_worker(app, perform_ping_check=False)
        cls.celery_worker.__enter__()

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

        # Close worker
        cls.celery_worker.__exit__(None, None, None)

    def test_my_function(self):
        # my_task.delay() or something

无论出于何种原因，测试工作者都会尝试使用名为'celery.ping' 的任务，可能是为了在工作者失败的情况下提供更好的错误消息。它正在寻找的任务是celery.contrib.testing.tasks.ping，在测试时不可用。将start_worker 的perform_ping_check 参数设置为False 会跳过对此的检查并避免相关的错误。

现在，运行测试时，无需启动单独的 Celery 进程。 Celery worker 将在 Django 测试过程中作为单独的线程启动。该工作人员可以查看任何内存数据库，包括默认的内存测试数据库。要控制工作人员的数量，start_worker 中提供了一些选项，但默认情况下似乎是单个工作人员。

【讨论】：

嗨，这种方法看起来确实很有趣——但是当我使用这种方法时，我似乎无法与 celery worker 交流。如果我有空闲时间，我会四处寻找。

解决 ping 任务问题的方法是在 cls.celery_worker = start_worker(app) 之前添加 app.loader.import_module('celery.contrib.testing.tasks') 或将 'celery.contrib.testing.tasks' 添加到 INSTALLED_APPS（注意末尾的 .tasks）。

如何读取/访问 celery worker 生成后创建的数据？它似乎适用于在 setUpClass 函数中创建的数据，但我得到了之后创建的任何数据的空数据。

【参考方案2】：

我根据@drhagen 的解决方案找到了另一种解决方法：

在拨打start_worker(app)之前先拨打celery.contrib.testing.app.TestApp()

from celery.contrib.testing.worker import start_worker
from celery.contrib.testing.app import TestApp

from myapp.tasks import app, my_task


class TestTasks:
    def setup(self):
        TestApp()
        self.celery_worker = start_worker(app)
        self.celery_worker.__enter__()

    def teardown(self):
        self.celery_worker.__exit__(None, None, None)

【讨论】：

我遇到了与django-channles 类似的问题，请问有什么帮助吗？ my question

【参考方案3】：

对于您的单元测试，我建议您跳过 celery 依赖项，以下两个链接将为您提供启动单元测试所需的信息：

http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.html http://docs.celeryproject.org/en/latest/userguide/testing.html

如果您真的想测试包括队列在内的 celery 函数调用，我可能会设置一个带有服务器、工作程序、队列组合的 dockercompose，并从 django-celery 文档中扩展自定义 CeleryTestRunner。但我看不出它有什么好处，因为测试系统离生产很远，无法具有代表性。

【讨论】：

他们实际上告诉你创建一个新的 Testrunner 并使用这个 testrunner 运行你的测试，导致测试跳过 celery 的异步部分。有了这个，您应该能够将您的测试视为同步，并导致（前）celery 任务的两个部分都命中同一个 testdb。

你说的是CELERY_ALWAYS_EAGER吗？我试过了。当我将其设置为 True 时，Celery 仍然使用生产数据库；它只是在 delay 返回之前等待任务完成。

嗯，很奇怪。您是否使用新的测试运行程序进行了测试，我一开始忘记了这一点，我花了一段时间才意识到我的错误^^

我们正在使用继承自 CeleryTestSuiteRunner 的 Testrunner。但我们正在做的唯一自定义事情是进行一些自定义夹具加载。

让我们continue this discussion in chat.