Тестирование Django Rest Framework с помощью очереди python

У меня есть простое приложение DRF с нативной очередью Python, для которого я сейчас пишу тесты. Один из API принимает файл, который затем помещается в очередь Python для обработки. Это прекрасно работает в приложении Django. Однако я не могу понять, как правильно протестировать это. Когда я вызываю API с помощью Django API Client, я вижу, что файл помещается в очередь, и начинается его обработка. Однако, когда ответ возвращается, тесты завершаются до того, как будет выполнена задача очереди. Затем эта задача так и не завершается. Я также не могу заставить тест заснуть, потому что это, похоже, приостанавливает и очередь. Что странно, потому что они должны быть в разных потоках.

Кроме того, по умолчанию tearDown приводит к ошибке ObjectInUse от psycopg2, так как тестовая БД используется. Я предполагаю, что это моя очередь обращается к БД, так как же мне правильно закрыть очередь, чтобы не произошло ошибки при разрыве? Мое предположение - закрыть поток, но класс Thread, похоже, не имеет такого метода.

Мой тест: 

from django.test import TestCase
from rest_framework.test import APIClient


class UploadTestCase(TestCase):

    def setUp(self):
        self.client = APIClient()
        self.client.force_authenticate()

    def test_post_file(self):
        with open('tmp.csv', 'rt') as f:
            response = self.client.post('/upload', data={'files': [f]})
            # Asserts
            ...

Прошу прощения, что это не "настоящий" ответ, но он стал длиннее, чем позволял комментарий.

Новый билет выглядит хорошо. Я заметил, что не было остановки фонового потока, как это сделали вы. Возможно, именно это и вызывает проблему с тем, что бд все еще активна.

Вы используете TestCase, который запускает транзакцию db и отменяет все изменения в базе данных после завершения тестовой функции. Это означает, что вы не сможете увидеть данные из тестового примера в другом потоке, использующем другое соединение с базой данных. Вы можете видеть их внутри ваших тестов и представлений, поскольку они используют общее соединение.

Celery и RQ являются стандартными очередями заданий - Celery более гибкая, а RQ более простая. Начните с RQ и сохраняйте все простым и изолированным.

Некоторые заметки:

  • Передавайте PK объектов, а не весь объект
    • .
  • Почитайте о pickle, если вам нужно передавать большие данные
    • .
  • Установите для queues значение async=False (выполняйте как обычный код) в tests.
    • .

Для QR я использую этот фрагмент для установки очередей в режим синхронизации в режиме тестирования, а затем просто запускаю все как обычно.

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

Удачи!

Вернуться на верх