使用 python 队列进行 Django Rest Framework 测试

Question

我有一个 DRF 应用程序，其中有一个 python 队列，我正在为其编写测试。不知何故，

我的队列线程找不到测试数据库中存在的对象。
主线程无法销毁数据库，因为其他 1 个会话正在使用它。

为了进一步解释用例，我使用了 Django 的用户模型，并且有一个 table 用于您可以上传的文件的元数据。其中一个字段是 created_by，它是 django.conf.settings.AUTH_USER_MODEL 的外键。如下所示，我在 TestCase 的 setUp() 中创建了一个用户，然后我使用它在文件 table 中创建了一个条目。但是，此条目的创建发生在队列中。在测试期间，这会导致错误 DETAIL: Key (created_by_id)=(4) is not present in table "auth_user".。测试完成后，tearDown 尝试销毁测试数据库，我收到另一个错误 DETAIL: There is 1 other session using the database.。两者似乎相关，我可能处理队列不正确。

测试是用 Django 的 TestCase 编写的，运行是用 python manage.py test。

from django.contrib.auth.models import User
from rest_framework.test import APIClient
from django.test import TestCase

class MyTest(TestCase):

    def setUp(self):
        self.client = APIClient()
        self.client.force_authenticate()
        user = User.objects.create_user('TestUser', 'test@test.test', 'testpass')
        self.client.force_authenticate(user)
   
    def test_failing(self):
        self.client.post('/totestapi', data={'files': [open('tmp.txt', 'rt')]})

队列在单独的文件中定义，app/queue.py。

from app.models import FileMeta
from queue import Queue
from threading import Thread


def queue_handler():
    while True:
        user, files = queue.get()
        for file in files:
            upload(file)
            FileMeta(user=user, filename=file.name).save()
        queue.task_done()


queue = Queue()
thread = Thread(target=queue_handler, daemon=True)

def start_upload_thread():
    thread.start()

def put_upload_thread(*args):
    queue.put(args)

最后，队列从app/views.py开始，Django启动时总是调用，包含所有API。

from rest_framework import APIView
from app.queue import start_upload_thread, put_upload_thread


start_upload_thread()

class ToTestAPI(APIView):

    def post(self, request):
        put_upload_thread(request.user, request.FILES.getlist('files'))

Answer 1

抱歉，这不是一个“真正的”答案，但它比评论允许的时间要长。

新票看起来不错。我确实注意到后台线程没有像您那样停止。这可能是导致数据库仍然处于活动状态的问题的原因。

您使用 TestCase，它运行是一个数据库事务，并在测试函数结束时撤消所有数据库更改。这意味着您将无法使用不同的数据库连接在另一个线程中查看来自测试用例的数据。您可以在您的测试和视图中看到它，因为它们共享一个连接。

Celery and RQ 是标准的作业队列——Celery 更灵活，但 RQ 更简单。从 RQ 开始，保持简单和独立。

一些注意事项：

传入对象的PK而不是整个对象
如果您确实需要传递更大的数据，请继续阅读 pickle。
在测试中设置 queues to async=False（运行就像普通代码一样）。

队列消费者是一个单独的进程运行在系统的任何地方，所以数据需要以某种方式到达他们。如果您使用完整对象，则需要对这些对象进行 pickled 或序列化，并保存在队列本身（即 redis）中以供检索和处理。请小心，不要以这种方式传递大对象 - 使用 PK，将文件存储在 S3 或其他对象存储中的某处，等等。

对于 Django-RQ，我在测试时使用此代码段将队列设置为同步模式，然后运行一切正常。

if IS_TESTING:
    for q in RQ_QUEUES.keys():
        RQ_QUEUES[q]['ASYNC'] = False

祝你好运！

使用 python 队列进行 Django Rest Framework 测试

Django Rest Framework testing with python queue

python

django

django-testing

django-rest-framework