使 Django 测试用例数据库对 Celery 可见 [英] Make Django test case database visible to Celery

问题描述

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

When a Django test case runs, it creates an isolated test database so that database writes get rolled back when each test completes. I am trying to create an integration test with Celery, but I can't figure out how to connect Celery to this ephemeral test database. In the naive setup, Objects saved in Django are invisible to Celery and objects saved in Celery persist indefinitely.

这是一个示例测试用例:

Here is an example test case:

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

我有两个问题:

1. 如何做到让 Celery 可以看到 Django 测试用例中的对象?

1. How do make it so that Celery can see the objects that the Django test case?

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

How do I ensure that all objects saved by Celery are automatically rolled back once the test completes?

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

I am willing to manually clean up the objects if doing this automatically is not possible, but a deletion of objects in tearDown even in APISimpleTestCase seems to be rolled back.

推荐答案

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

This is possible by starting a Celery worker within the Django test case.

Django 的内存数据库是 sqlite3.正如 Sqlite 内存数据库的描述页面 上所说，[A]ll 数据库连接共享内存数据库需要在同一个进程中."这意味着，只要 Django 使用内存中的测试数据库，并且 Celery 是在单独的进程中启动的，那么 Celery 和 Django 共享一个测试数据库从根本上是不可能的.

Django's in-memory database is sqlite3. As it says on the description page for Sqlite in-memory databases, "[A]ll database connections sharing the in-memory database need to be in the same process." This means that, as long as Django uses an in-memory test database and Celery is started in a separate process, it is fundamentally impossible to have Celery and Django to share a test database.

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

However, with celery.contrib.testing.worker.start_worker, it possible to start a Celery worker in a separate thread within the same process. This worker can access the in-memory database.

这假设 Celery 已经在 与 Django 项目的通常方式.

This assumes that Celery is already setup in the usual way with the Django project.

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

Because Django-Celery involves some cross-thread communication, only test cases that don't run in isolated transactions will work. The test case must inherit directly from SimpleTestCase or its Rest equivalent APISimpleTestCase and set databases to '__all__' or just the database that the test interacts with.

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

The key is to start a Celery worker in the setUpClass method of the TestCase and close it in the tearDownClass method. The key function is celery.contrib.testing.worker.start_worker, which requires an instance of the current Celery app, presumably obtained from mysite.celery.app and returns a Python ContextManager, which has __enter__ and __exit__ methods, which must be called in setUpClass and tearDownClass, respectively. There is probably a way to avoid manually entering and existing the ContextManager with a decorator or something, but I couldn't figure it out. Here is an example tests.py file:

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 可跳过此检查并避免相关错误.

For whatever reason, the testing worker tries to use a task called 'celery.ping', probably to provide better error messages in the case of worker failure. The task it is looking for is celery.contrib.testing.tasks.ping, which is not available at test time. Setting the perform_ping_check argument of start_worker to False skips the check for this and avoids the associated error.

现在，当运行测试时，无需启动单独的 Celery 进程.Celery worker 将在 Django 测试进程中作为一个单独的线程启动.该工作者可以查看任何内存数据库，包括默认的内存测试数据库.为了控制worker的数量，start_worker中有可用的选项，但似乎默认是单个worker.

Now, when the tests are run, there is no need to start a separate Celery process. A Celery worker will be started in the Django test process as a separate thread. This worker can see any in-memory databases, including the default in-memory test database. To control the number of workers, there are options available in start_worker, but it appears the default is a single worker.

这篇关于使 Django 测试用例数据库对 Celery 可见的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！