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

Question

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

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

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

测试是用 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', '<a href="https://stackoverflow.com/cdn-cgi/l/email-protection" class="__cf_email__" data-cfemail="473322343307332234336933223433" rel="noreferrer noopener nofollow">[email protected]</a>', '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 ，它运行数据库事务并在测试函数结束时撤消所有数据库更改。这意味着您将无法使用与数据库的不同连接在另一个线程中查看测试用例中的数据。您可以在测试和 View 中看到它，因为它们共享连接。

Celery和 RQ是标准作业队列 - Celery 更灵活，但 RQ 更简单。从 RQ 开始，让事情变得简单和隔离。

一些注意事项:

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

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

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

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

祝你好运!