我有一个DRF应用程序，其中包含一个python队列，我正在为其编写测试。不知何故
- 我的队列线程找不到测试数据库中存在的对象。
- 主线程无法销毁数据库，因为它正被其他 1 个会话使用。
为了进一步解释用例，我使用了 Django 的用户模型，并有一个可以上传的文件元数据表。其中一个字段是created_by，它是django.conf.settings.AUTH_USER_MODEL的外键。如下所示，我在测试用例的setUp()中创建了一个用户，然后使用它在文件表中创建一个条目。但是，此条目的创建发生在队列中。在测试期间，这会导致错误DETAIL: Key (created_by_id)=(4) is not present in table "auth_user".。 当测试完成并且拆解试图破坏测试数据库时，我收到另一个错误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'))
抱歉，这不是一个"真正的"答案，但它比评论允许的要长。
新票看起来不错。 我确实注意到后台线程没有像您一样停止。 这可能是导致数据库仍处于活动状态的问题的原因。
您可以使用 TestCase，它运行数据库事务并在测试函数结束时撤消所有数据库更改。 这意味着您将无法使用与数据库的不同连接在另一个线程中查看测试用例中的数据。 您可以在测试和视图中看到它，因为它们共享连接。
芹菜和 RQ 是标准的作业队列 - 芹菜更灵活，但 RQ 更简单。 从 RQ 开始，保持简单和隔离。
一些注意事项：
- 传入对象的PK而不是整个对象
- 如果您确实需要传递更大的数据，请阅读泡菜。
- 在测试中将队列设置为 async=False(像普通代码一样运行)。
队列使用者是一个单独的进程，在系统中的任何地方运行，因此数据需要以某种方式到达它们。 如果您使用完整的对象，则需要pickled或序列化这些对象，并将其保存在队列本身(即 redis)中以进行检索和处理。 请注意，不要以这种方式传递大型对象 - 使用 PK，将文件存储在 S3 或其他对象存储中的某个位置等。
对于 Django-RQ，我使用此代码片段在测试时将队列设置为同步模式，然后正常运行。
if IS_TESTING:
for q in RQ_QUEUES.keys():
RQ_QUEUES[q]['ASYNC'] = False
祝你好运！