Unidad de pruebas de un asíncrono en apio.

Estoy tratando de probar alguna funcionalidad de apio en el marco de pruebas de unidades de Django, pero cada vez que bash verificar un AsyncResult, las pruebas actúan como si nunca hubieran comenzado.

Sé que este código funciona en un entorno real con RabbitMQ, así que me preguntaba por qué no funcionaba al usar el marco de prueba.

Aquí hay un ejemplo:

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

¿El uso de la opción ALWAYS_EAGER deshabilita la funcionalidad AsyncResult ya que se ejecuta de inmediato? Si es así, ¿hay alguna manera de poder realizar una prueba unitaria de la comprobación de estado de AsyncResult? Si trato de eliminar la opción ALWAYS_EAGER, las pruebas nunca se ejecutan, por lo que estoy perdido.

¡Gracias!

Cuando CELERY_ALWAYS_EAGER es True , la llamada a apply_async() realidad se reemplaza con apply() . El resultado devuelto es un EagerResult , que ya contiene el resultado de su tarea.

Entonces, sí, la configuración de ALWAYS_EAGER = True deshabilita toda la funcionalidad de AsyncResult . Se omite todo el proceso asíncrono y, en realidad, no se envía ninguna tarea al intermediario, por lo que no puede recuperar el resultado a través de un AsyncResult .

Use CELERY_ALWAYS_EAGER = True cuando está probando la ruta del código que solo necesita un resultado de apio, y funciona de la misma manera con un EagerResult o un AsyncResult .

Si es necesario, también hay una manera de ejecutar pruebas con AsyncResult , con CELERY_ALWAYS_EAGER = False , pero para esto, deberá comenzar un trabajador antes de llamar a la tarea en su caso de prueba. El trabajador podrá ejecutar su tarea y AsyncResult funcionará bien. Puedes echarle un vistazo a django-celery-testworker, que parece hacer precisamente eso, aunque no lo he probado.