In a recent project, I used Django with Celery. When testing, I ran into some problem and it took me a while to solve. Hope this can help you a bit if you see the same thing.
In my project, I tried to use Celery to send email. Meanwhile, I have some model called EmailLog to track the emails. Part of the code is like this:
send_my_email function, I created an EmailLog object, saved it first and then send the email using Celery. In the Celery task, the email_log’s status will be updated. The two print statements ensure that
email_log is actually in the database.
While running unittest, however, my Celery worker can’t find the saved EmailLog object, saying ‘matching query does not exist’. It seems that the tests are wrapped in some transaction and even if both Django and Celery are configured to use the same database, they can’t share data.
Finally, I set
CELERY_ALWAYS_EAGER=True to bypass this problem. It will execute the task locally, not in the queue. At least it can verify the correctness of the function.