Make Celery use Django's test database without task_always_eager

Question

When running tests in Django applications that make use of Celery tasks I can't fully test tasks that need to get data from the database since they don't connect to the test database that Django creates.

Setting task_always_eager in Celery to True partially solves this problem but as the documentation for testing says, this doesn't fully reflect how the code will run on a real Celery worker and isn't suitable for testing.

How can I make Celery tasks use the Django test database when running Django tests without setting task_always_eager = True?

Answer 1

Short = You must run celery worker as in production

Easy:

1. Use dedicated test db (as in production)
2. Configure celery to use it
3. Start celery worker manually before you run tests

Advanced:

1. Use auto created test db (it may be sqlite)
2. Run celery worker in your test setUp()
3. Configure celery to use auto created test db (copy django.conf.settings.DATABASE from test process to celery)

And always you must provide message broker for celery.

I have a test that requires dedicated celery worker to check my code that passes messages between celery task and calling code: https://gist.github.com/Sovetnikov/a7ad982fc77e8dfbc528bfc20fcf3b1e This python module is two in one - a TestUnit and celery worker runner with self contained configuration.

My code does not utilize any db, but you can easily adapt it to your need. Just pass django.conf.settings.DATABASE (as json or pickle or whatever you like method) to celery starter code and configure Django DATABASE to point to test db.

Additional info:

1. There is complete solution for this case https://github.com/RentMethod/celerytest (i tried some old version of it and have no luck because it uses threads, with python GIL ... and i think it is over-complicated)

2. Sample code, how to configure DATABASE settings and init django itself in single module https://gist.github.com/Sovetnikov/369a8d05ba2b6482fa20769bc498f122

Answer 2

A simple solution is to use celery.contrib.testing.worker.start_worker to spawn a Celery worker within the Django test process. Because it lives in the same process, it can access the default in-memory test database, but because it lives in its own thread, it isn't eager and the task_always_eager flag is not needed or recommended.