इकाई सेलेरी में एक AsyncResult परीक्षण

Question

मैं Django के यूनिट परीक्षण ढांचे में कुछ अजवाइन कार्यक्षमता का परीक्षण करने की कोशिश कर रहा हूं, लेकिन जब भी मैं AsyncResult की जांच करने की कोशिश करता हूं तो परीक्षण कार्य शुरू नहीं होता था।

मुझे पता है कि यह कोड खरगोश एमक्यू के साथ एक वास्तविक वातावरण में काम करता है, इसलिए मैं सोच रहा था कि यह परीक्षण ढांचे का उपयोग करते समय क्यों काम नहीं करता था।

यहाँ एक उदाहरण है:

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

के बाद से इसे तुरंत क्रियान्वित है ALWAYS_EAGER विकल्प AsyncResult कार्यक्षमता को अक्षम का उपयोग कर रहा है? यदि हां, तो क्या यूनिट परीक्षण AsyncResult स्थिति जांच करने में सक्षम करने का कोई तरीका है? यदि मैं ALWAYS_EAGER विकल्प को निकालने का प्रयास करता हूं तो परीक्षण कभी नहीं चलते हैं, इसलिए मुझे नुकसान हुआ है।

धन्यवाद!

Answer 1

CELERY_ALWAYS_EAGERTrue है, apply_async() पर कॉल वास्तव में apply() के साथ प्रतिस्थापित किया गया है। परिणाम लौटा एक EagerResult है, जिसमें पहले से ही आपके कार्य का परिणाम शामिल है।

तो, हाँ, ALWAYS_EAGER = True सेटिंग AsyncResult कार्यक्षमता अक्षम करता है। संपूर्ण एसिंक प्रक्रिया को छोड़ दिया गया है, और वास्तव में ब्रोकर को कोई कार्य नहीं भेजा जाता है, यही कारण है कि आप AsyncResult के माध्यम से परिणाम पुनर्प्राप्त नहीं कर सकते हैं।

CELERY_ALWAYS_EAGER = True का उपयोग करें जब आप कोड पथ का परीक्षण कर रहे हैं, जिसे केवल सेलेरी परिणाम की आवश्यकता है, और EagerResult या AsyncResult के साथ उसी तरह काम करें।

यदि आवश्यक हो, तो AsyncResult के साथ CELERY_ALWAYS_EAGER = False के साथ परीक्षण चलाने का एक तरीका है, लेकिन इसके लिए, आपको अपने परीक्षण मामले में कार्य को कॉल करने से पहले एक कार्यकर्ता शुरू करना होगा। कार्यकर्ता तब आपके कार्य को निष्पादित करने में सक्षम होगा, और AsyncResult ठीक काम करेगा। आप django-celery-testworker पर एक नज़र डाल सकते हैं जो ऐसा लगता है, हालांकि मैंने इसका परीक्षण नहीं किया है।