Đơn vị kiểm tra AsyncResult trong cần tây

Question

Tôi đang thử kiểm tra một số chức năng cần tây trong khung kiểm thử đơn vị của Django, nhưng bất cứ khi nào tôi thử kiểm tra AsyncResult thì các thử nghiệm hành động như nó chưa bao giờ bắt đầu.

Tôi biết mã này hoạt động trong môi trường thực tế với RabbitMQ, vì vậy tôi đã tự hỏi tại sao nó không hoạt động khi sử dụng khung kiểm tra.

Dưới đây là một ví dụ:

@override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS = True, 
        CELERY_ALWAYS_EAGER = True, 
        BROKER_BACKEND = 'memory',) 
def test_celery_do_work(self): 
    result = myapp.tasks.celery_do_work.AsyncResult('blat') 
    applied_task = myapp.tasks.celery_do_work.apply_async((), task_id='blat') 
    applied_task.wait() 
    # THIS SUCCEEDS 
    self.assertTrue(applied_task.successful()) 
    # THIS FAILS 
    self.assertTrue(result.successful()) 

Không sử dụng tùy chọn ALWAYS_EAGER vô hiệu hóa chức năng asyncResult kể từ khi nó được thực hiện ngay lập tức? Nếu vậy, có cách nào để có thể kiểm tra trạng thái kiểm tra trạng thái AsyncResult không? Nếu tôi cố gắng đưa ra lựa chọn ALWAYS_EAGER thì các bài kiểm tra không bao giờ chạy, vì vậy tôi thua lỗ.

Cảm ơn!

Answer 1

Khi CELERY_ALWAYS_EAGER là True, cuộc gọi đến apply_async() thực sự được thay thế bằng apply(). Kết quả trả về là EagerResult, kết quả của tác vụ của bạn.

Vì vậy, có, đặt ALWAYS_EAGER = True sẽ tắt toàn bộ chức năng của AsyncResult. Toàn bộ quá trình async bị bỏ qua, và không có nhiệm vụ nào thực sự được gửi tới người môi giới, đó là lý do tại sao bạn không thể truy xuất kết quả thông qua một số AsyncResult.

Sử dụng CELERY_ALWAYS_EAGER = True khi bạn đang thử nghiệm đường dẫn mã chỉ cần kết quả Celery và làm việc theo cách tương tự với EagerResult hoặc AsyncResult.

Nếu cần, có một cách để chạy thử nghiệm với AsyncResult quá, với CELERY_ALWAYS_EAGER = False, nhưng đối với điều này, bạn sẽ cần phải bắt đầu một nhân viên trước khi gọi nhiệm vụ trong trường hợp thử nghiệm của bạn. Sau đó, nhân viên sẽ có thể thực hiện nhiệm vụ của bạn và AsyncResult sẽ hoạt động tốt. Bạn có thể xem django-celery-testworker mà dường như làm điều đó, mặc dù tôi đã không thử nghiệm nó.