Thử nghiệm đơn vị với django-cần tây?

Question

Tôi đang cố gắng đưa ra phương pháp thử nghiệm cho dự án django-celery của chúng tôi. Tôi đã đọc các ghi chú trong các documentation, nhưng nó đã không cho tôi một ý tưởng tốt về những gì để thực sự làm. Tôi không lo lắng về việc thử nghiệm các tác vụ trong các daemon thực, chỉ cần chức năng của mã số của tôi. Chủ yếu tôi tự hỏi:

1. Làm cách nào chúng tôi có thể bỏ qua task.delay() trong khi thử nghiệm (Tôi đã thử đặt CELERY_ALWAYS_EAGER = True nhưng không có sự khác biệt)?
2. Làm cách nào để sử dụng cài đặt thử nghiệm được đề xuất (nếu đó là cách tốt nhất) mà không thực sự thay đổi settings.py của chúng tôi?
3. Chúng tôi vẫn có thể sử dụng manage.py test hoặc chúng tôi có phải sử dụng nhân viên tùy chỉnh không?

Nhìn chung, mọi gợi ý hoặc mẹo để thử nghiệm cần tây sẽ rất hữu ích.

Answer 1

Thử cài đặt:

BROKER_BACKEND = 'memory' 

(Nhờ bình luận asksol 's.)

Answer 2

Dưới đây là một đoạn trích từ lớp cơ sở thử nghiệm của tôi để khai báo phương pháp apply_async và ghi lại các cuộc gọi đến nó (bao gồm Task.delay.) Đó là một chút, nhưng nó được quản lý để phù hợp với nhu cầu của tôi trong vài tháng qua ' đã sử dụng nó.

from django.test import TestCase 
from celery.task.base import Task 
# For recent versions, Task has been moved to celery.task.app: 
# from celery.app.task import Task 
# See http://docs.celeryproject.org/en/latest/reference/celery.app.task.html 

class CeleryTestCaseBase(TestCase): 

    def setUp(self): 
     super(CeleryTestCaseBase, self).setUp() 
     self.applied_tasks = [] 

     self.task_apply_async_orig = Task.apply_async 

     @classmethod 
     def new_apply_async(task_class, args=None, kwargs=None, **options): 
      self.handle_apply_async(task_class, args, kwargs, **options) 

     # monkey patch the regular apply_sync with our method 
     Task.apply_async = new_apply_async 

    def tearDown(self): 
     super(CeleryTestCaseBase, self).tearDown() 

     # Reset the monkey patch to the original method 
     Task.apply_async = self.task_apply_async_orig 

    def handle_apply_async(self, task_class, args=None, kwargs=None, **options): 
     self.applied_tasks.append((task_class, tuple(args), kwargs)) 

    def assert_task_sent(self, task_class, *args, **kwargs): 
     was_sent = any(task_class == task[0] and args == task[1] and kwargs == task[2] 
         for task in self.applied_tasks) 
     self.assertTrue(was_sent, 'Task not called w/class %s and args %s' % (task_class, args)) 

    def assert_task_not_sent(self, task_class): 
     was_sent = any(task_class == task[0] for task in self.applied_tasks) 
     self.assertFalse(was_sent, 'Task was not expected to be called, but was. Applied tasks: %s' %     self.applied_tasks) 

Dưới đây là một "ra khỏi đỉnh đầu" dụ về cách bạn sẽ sử dụng nó trong trường hợp thử nghiệm của bạn:

mymodule.py

from my_tasks import SomeTask 

def run_some_task(should_run): 
    if should_run: 
     SomeTask.delay(1, some_kwarg=2) 

test_mymodule.py

class RunSomeTaskTest(CeleryTestCaseBase): 
    def test_should_run(self): 
     run_some_task(should_run=True) 
     self.assert_task_sent(SomeTask, 1, some_kwarg=2) 

    def test_should_not_run(self): 
     run_some_task(should_run=False) 
     self.assert_task_not_sent(SomeTask) 

Answer 3

Tôi thích sử dụng các override_settings trang trí trên các bài kiểm tra mà cần kết quả cần tây để hoàn thành.

from django.test import TestCase 
from django.test.utils import override_settings 
from myapp.tasks import mytask 

class AddTestCase(TestCase): 

    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True, 
         CELERY_ALWAYS_EAGER=True, 
         BROKER_BACKEND='memory') 
    def test_mytask(self): 
     result = mytask.delay() 
     self.assertTrue(result.successful()) 

Nếu bạn muốn áp dụng điều này cho tất cả các bài kiểm tra, bạn có thể sử dụng các Á hậu kiểm tra cần tây như đã mô tả ở http://docs.celeryproject.org/en/2.5/django/unit-testing.html mà về cơ bản đặt những thiết lập tương tự trừ (BROKER_BACKEND = 'memory').

Trong cài đặt:

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner' 

Nhìn vào nguồn cho CeleryTestSuiteRunner và nó khá rõ ràng những gì đang xảy ra.

Answer 4

kể từ khi tôi vẫn thấy điều này đưa ra trong kết quả tìm kiếm, cài đặt ghi đè với

TEST_RUNNER = 'djcelery.contrib.test_runner.CeleryTestSuiteRunner' 

làm việc cho tôi theo Celery Docs