Лучшая практика тестирования django-rq (python-rq) в Django

Question

Я начну использовать django-rq в своем проекте.

Интеграция Django с RQ, основанной на Redis библиотеке очередей Python.

Какова наилучшая практика тестирования приложений django, использующих RQ?

Например, если я хочу проверить свое приложение как черный ящик, после того, как Пользователь выполнит некоторые действия, я хочу выполнить все задания в текущей очереди, а затем проверить все результаты в моей БД. Как я могу это сделать в своих джанго-тестах?

Answer 1

Вам понадобятся ваши тесты для паузы, пока в очереди еще есть задания. Чтобы сделать это, вы можете проверить Queue.is_empty() и приостановить выполнение, если все еще есть рабочие места в очереди:

import time 
from django.utils.unittest import TestCase 
import django_rq 

class TestQueue(TestCase): 

def test_something(self): 
    # simulate some User actions which will queue up some tasks 

    # Wait for the queued tasks to run 
    queue = django_rq.get_queue('default') 
    while not queue.is_empty(): 
     time.sleep(5) # adjust this depending on how long your tasks take to execute 

    # queued tasks are done, check state of the DB 
    self.assert(.....) 

Answer 2

Я только что нашел django-rq, что позволяет раскручивать работник в тестовой среде, которая выполняет любые задачи по очередь, а затем завершает работу.

from django.test impor TestCase 
from django_rq import get_worker 

class MyTest(TestCase): 
    def test_something_that_creates_jobs(self): 
     ...      # Stuff that init jobs. 
     get_worker().work(burst=True) # Processes all jobs then stop. 
     ...      # Asserts that the job stuff is done. 

Answer 3

Я отделил свои rq тестов на несколько частей.

1. Проверьте, что я правильно добавляю вещи в очередь (используя mocks).
2. Предположим, что если что-то добавится в очередь, оно в конечном итоге будет обработано. (rq - это набор тестов).
3. Испытание, учитывая правильный ввод, мои задачи работают должным образом. (обычные тесты кода).

Код тестируется:

def handle(self, *args, **options): 
    uid = options.get('user_id') 

    # @@@ Need to exclude out users who have gotten an email within $window 
    # days. 
    if uid is None: 
     uids = User.objects.filter(is_active=True, userprofile__waitlisted=False).values_list('id', flat=True) 
    else: 
     uids = [uid] 

    q = rq.Queue(connection=redis.Redis()) 

    for user_id in uids: 
     q.enqueue(mail_user, user_id) 

Мои тесты:

class DjangoMailUsersTest(DjangoTestCase): 
    def setUp(self): 
     self.cmd = MailUserCommand() 

    @patch('redis.Redis') 
    @patch('rq.Queue') 
    def test_no_userid_queues_all_userids(self, queue, _): 
     u1 = UserF.create(userprofile__waitlisted=False) 
     u2 = UserF.create(userprofile__waitlisted=False) 
     self.cmd.handle() 
     self.assertItemsEqual(queue.return_value.enqueue.mock_calls, 
           [call(ANY, u1.pk), call(ANY, u2.pk)]) 

    @patch('redis.Redis') 
    @patch('rq.Queue') 
    def test_waitlisted_people_excluded(self, queue, _): 
     u1 = UserF.create(userprofile__waitlisted=False) 
     UserF.create(userprofile__waitlisted=True) 
     self.cmd.handle() 
     self.assertItemsEqual(queue.return_value.enqueue.mock_calls, [call(ANY, u1.pk)]) 

Answer 4

Я совершил a patch, что позволяет сделать:

from django.test impor TestCase 
from django_rq import get_queue 

class MyTest(TestCase): 
    def test_something_that_creates_jobs(self): 
     queue = get_queue(async=False) 
     queue.enqueue(func) # func will be executed right away 
     # Test for job completion 

Это должно сделать тестирование RQ работу проще , Надеюсь, это поможет!

Answer 5

На всякий случай это было бы полезно для всех. Я использовал патч с пользовательским фиктивным объектом, чтобы сделать Епдиеее что бы работать сразу

#patch django_rq.get_queue 
with patch('django_rq.get_queue', return_value=MockBulkJobGetQueue()) as mock_django_rq_get_queue: 
    #Perform web operation that starts job. In my case a post to a url 

Тогда фиктивный объект просто был один метод:

class MockBulkJobGetQueue(object): 

    def enqueue(self, f, *args, **kwargs): 
     # Call the function 
     f(
      **kwargs.pop('kwargs', None) 
     ) 

Answer 6

, что я сделал для этого случая определить, тестирую ли я, и использовать fakeredis во время тестов. наконец, в самом тесте, я епдиеие в Redis уборщица задачу в режиме синхронизирующего:

первый, определить функцию, которая определяет, если вы тестируете:

TESTING = len(sys.argv) > 1 and sys.argv[1] == 'test' 

def am_testing(): 
    return TESTING 

затем в файле, который использует Redis в очередь таким образом, управлять очередью. вы можете расширить get_queue указать имя очереди, если это необходимо:

if am_testing(): 
    from fakeredis import FakeStrictRedis 
    from rq import Queue 
    def get_queue(): 
     return Queue(connection=FakeStrictRedis()) 

else: 
    import django_rq 
    def get_queue(): 
     return django_rq.get_queue() 

тогда, епдиеие свою задачу следующим образом:

queue = get_queue() 
queue.enqueue(task_mytask, arg1, arg2) 

наконец, в тестовой программе, выполнить задание выполняется тестирование в синхронизации так что он работает в том же процессе, что и ваш тест. Как показывает практика, я первый очистить очередь fakeredis, но я не думаю, что его необходимо, так как нет рабочих:

from rq import Queue 
from fakeredis import FakeStrictRedis 

FakeStrictRedis().flushall() 
queue = Queue(async=False, connection=FakeStrictRedis()) 
queue.enqueue(task_mytask, arg1, arg2) 

мой settings.py имеет нормальные параметры django_redis, так django_rq.getqueue() использует их при развертывании:

RQ_QUEUES = { 
    'default': { 
     'HOST': env_var('REDIS_HOST'), 
     'PORT': 6379, 
     'DB': 0, 
     # 'PASSWORD': 'some-password', 
     'DEFAULT_TIMEOUT': 360, 
    }, 
    'high': { 
     'HOST': env_var('REDIS_HOST'), 
     'PORT': 6379, 
     'DB': 0, 
     'DEFAULT_TIMEOUT': 500, 
    }, 
    'low': { 
     'HOST': env_var('REDIS_HOST'), 
     'PORT': 6379, 
     'DB': 0, 
    } 
}