当 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
中删除对象似乎会回滚。
最佳答案
这可以通过在 Django 测试用例中启动 Celery worker 来实现。
背景
Django 的内存数据库是sqlite3。正如它在 the description page for Sqlite in-memory databases 上所说, “[A] 所有共享内存数据库的数据库连接都需要在同一个进程中。”这意味着,只要Django使用内存中的测试数据库,Celery在单独的进程中启动,Celery和Django共享一个测试数据库是根本不可能的。
但是,对于 celery.contrib.testing.worker.start_worker
,可以在同一进程中的单独线程中启动 Celery worker。该工作人员可以访问内存中的数据库。
这假设 Celery 已经在 the usual way 中设置好了与 Django 项目。
解决方案
因为 Django-Celery 涉及一些跨线程通信,所以只有不在隔离事务中运行的测试用例才会起作用。测试用例必须直接继承自SimpleTestCase
或其其余等效项 APISimpleTestCase
并设置 databases
至 '__all__'
或者只是测试与之交互的数据库。
关键是在setUpClass
中启动一个Celery worker TestCase
的方法并在 tearDownClass
中关闭它方法。关键功能是celery.contrib.testing.worker.start_worker
,这需要当前 Celery 应用程序的实例,大概是从 mysite.celery.app
获得的并返回一个 Python ContextManager
, 其中有 __enter__
和 __exit__
方法,必须在 setUpClass
中调用和 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'
的任务,可能是为了在 worker 失败的情况下提供更好的错误消息。它正在寻找的任务是celery.contrib.testing.tasks.ping
,在测试时不可用。设置 perform_ping_check
start_worker
的参数至 False
跳过对此的检查并避免相关的错误。
现在,当测试运行时,无需启动单独的 Celery 进程。 Celery worker 将作为一个单独的线程在 Django 测试进程中启动。这个工作人员可以看到任何内存数据库,包括默认的内存测试数据库。要控制 worker 数量,start_worker
中提供了选项。 , 但默认情况下似乎是一个 worker 。
关于django - 使 Django 测试用例数据库对 Celery 可见,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/46530784/