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

【问题标题】：Django Rest Framework testing with python queue使用 python 队列进行 Django Rest Framework 测试
【发布时间】：2022-02-08 23:48:10
【问题描述】：

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

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

为了进一步解释这个用例，我使用了 Django 的用户模型，并有一个表，用于存储您可以上传的文件元数据。其中一个字段是created_by，它是django.conf.settings.AUTH_USER_MODEL 的ForeignKey。如下所示，我在TestCase 的setUp() 中创建了一个用户，然后我用它在Files 表中创建了一个条目。但是，此条目的创建发生在队列中。在测试期间，这会导致错误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'))

【问题讨论】：

对我来说这里发生的事情太多了。哪个测试运行器？什么是“python 原生队列”？（是 RQ 吗？）它是如何在测试中启动和停止的？您如何管理交易？ #1）你的消费者看不到你的测试用例所做的工作，#2）你的消费者在测试完成后没有被正确处理。对于＃1 - 您的测试用例是否提交了更改？消费者是否打开了新的 txn 以供阅读？对于＃2 - 消费者如何启动和停止？我们为单元测试运行同步队列，并为异步进行单独的集成测试。
为草率的票道歉。直到我读了你的评论才真正意识到。我编辑了代码，我使用了队列和线程库，当测试运行程序解析app/views.py 时开始发生，坦率地说，我没有停止队列，这可能是#2 的原因。但是，我认为我无法停止线程/队列。 AFAIK，测试用例正在提交，当我查询用户时，我找到了我创建的用户。什么是txn？此时我的猜测是我应该切换到 RQ/Django-Q

标签： python django django-rest-framework django-testing

【解决方案1】：

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

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

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

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

一些注意事项：

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

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

对于 Django-RQ，我在测试时使用这个 sn-p 将队列设置为同步模式，然后正常运行。

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

祝你好运！

【讨论】：

谢谢，这澄清了！