Thực hành tốt nhất kiểm tra django-rq (python-rq) ở Django

Question

Tôi sẽ bắt đầu sử dụng django-rq trong dự án của tôi.

Tích hợp Django với RQ, thư viện xếp hàng Python dựa trên Redis.

Cách tốt nhất để thử nghiệm các ứng dụng django đang sử dụng RQ là gì? Ví dụ: nếu tôi muốn thử nghiệm ứng dụng của mình dưới dạng hộp đen, sau khi Người dùng thực hiện một số hành động, tôi muốn thực hiện tất cả công việc trong Hàng đợi hiện tại, sau đó kiểm tra tất cả kết quả trong DB của tôi. Làm thế nào tôi có thể làm điều đó trong các bài kiểm tra django của tôi?

Answer 1

Bạn sẽ cần các bài kiểm tra của mình để tạm dừng trong khi vẫn còn các công việc trong hàng đợi. Để làm điều này, bạn có thể kiểm tra Queue.is_empty(), và đình chỉ thực hiện nếu vẫn còn những công việc trong hàng đợi:

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

Tôi chỉ tìm thấy django-rq, cho phép bạn quay về một người lao động trong một môi trường thử nghiệm mà thực hiện bất kỳ nhiệm vụ trên hàng đợi và sau đó thoát.

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

Tôi đã tách riêng các thử nghiệm rq thành một vài phần.

1. Kiểm tra xem tôi có đang thêm đúng thứ vào hàng đợi hay không (sử dụng mocks).
2. Giả sử rằng nếu một thứ gì đó được thêm vào hàng đợi thì nó sẽ được xử lý. (rq 's bộ thử nghiệm nên bao gồm điều này).
3. Kiểm tra, với đầu vào chính xác, nhiệm vụ của tôi hoạt động như mong đợi. (kiểm tra mã bình thường).

Mã đang được thử nghiệm:

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) 

xét nghiệm của tôi:

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

tôi cam kết a patch cho phép bạn làm:

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 

này nên thực hiện kiểm tra việc RQ dễ dàng hơn . Hy vọng rằng sẽ giúp!

Answer 5

Chỉ trong trường hợp điều này sẽ hữu ích cho bất kỳ ai. Tôi sử dụng một bản vá với một đối tượng tùy chỉnh mô hình để làm enqueue mà có thể chạy ngay lập tức

#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 

Sau đó, đối tượng giả chỉ có một phương pháp:

class MockBulkJobGetQueue(object): 

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

Answer 6

những gì tôi đã làm cho trường hợp này là để phát hiện nếu tôi đang thử nghiệm và sử dụng fakeredis trong khi kiểm tra. Cuối cùng, trong các thử nghiệm bản thân, tôi enqueue nhiệm vụ lao động redis trong chế độ đồng bộ:

đầu tiên, xác định một chức năng phát hiện nếu bạn đang thử nghiệm:

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

def am_testing(): 
    return TESTING 

sau đó trong tập tin của bạn có sử dụng redis phải xếp hàng lập nhiệm vụ, quản lý hàng đợi theo cách này. bạn có thể mở rộng get_queue để chỉ định tên hàng đợi nếu cần thiết:

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() 

sau đó, enqueue nhiệm vụ của bạn như sau:

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

cuối cùng, trong chương trình thử nghiệm của bạn, chạy về công việc bạn đang thử nghiệm đồng bộ chế độ, để nó chạy trong cùng một quá trình như thử nghiệm của bạn. Như một vấn đề của thực tế, đầu tiên tôi xóa hàng đợi fakeredis, nhưng tôi không nghĩ rằng cần thiết của nó vì không có người lao động:

from rq import Queue 
from fakeredis import FakeStrictRedis 

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

settings.py của tôi có các thiết lập django_redis bình thường, vì vậy django_rq.getqueue() sử dụng chúng khi được triển khai:

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, 
    } 
}