test_celery/run_tasks.py

   1 from .tasks import longtime_add
   2 import time
   3
   4 if __name__ == '__main__':
   5     result = longtime_add.delay(1,2)
   6     # at this time, our task is not finished, so it will return False
   7     print ('Task finished? ', result.ready())
   8     print ('Task result: ', result.result)
   9     # sleep 10 seconds to ensure the task has been finished
  10     time.sleep(10)
  11     # now the task should be finished and ready method will return True
  12     print ('Task finished? ', result.ready())
  13     print ('Task result: ', result.result)