Django TransactionTestCase and django-q sync

Django provides 2 base class that you can use to write tests for your django app - TestCase and TransactionTestCase. Most of the time TestCase is what you will be used. It's faster as it runs each test within a transaction and rollback the transaction when the test is finished.

The problem is that, if your code under test also need to test the effect of using transaction then you can't use TestCase. Django provides TransactionTestCase for this purpose. In this class, each test will have database setup and teardown process - tables created and migrations runs at the beginning of the test and being dropped when the test finish. This is applied to each test case so you can imagine how slow it is.

For more information you can refer to the documentation page.

django-q

Django-q is a lightweight task queue that we usually use in our Django project. We prefer it compared to celery and it support a number of brokers such as redis, SQS, IronMQ and RDBMS (through Django ORM).

One feature django-q provides is the sync parameter when adding task to the queue. The parameter is meant to bypass the queue and immediately executing the task function, which is something you usually do in tests.

In the beginning the sync implementation is quite simple, basically just calling the task function, something like:-

if sync:
    run(task_func, *args, **kwargs, ...)

But in recent implementation this has been changed so that it better simulates the environment when the task was executed within the queue framework. So the implementation is something likes:-

if sync:
    _sync(task)

and the _sync() function defined as:-

def _sync(pack):
    """Simulate a package travelling through the cluster."""
    from django_q.cluster import worker, monitor

    task_queue = Queue()
    result_queue = Queue()
    task = SignedPackage.loads(pack)
    task_queue.put(task)
    task_queue.put("STOP")
    worker(task_queue, result_queue, Value("f", -1))
    result_queue.put("STOP")
    monitor(result_queue)
    task_queue.close()
    task_queue.join_thread()
    result_queue.close()
    result_queue.join_thread()
    return task["id"]

So in the original implementation the task function simply executed, no different than just calling the function, which mean it still executed within the same process as the tests. All is well here but in reality, this is not how the task function will be executed. So the _sync() function above would simulate how the task would enter and popped out of the queue to be executed.

The side effect is that the current transaction also will be closed and remember again how django TestCase works? Long story shot our tests will break. So when writing tests that need to utilized the sync parameter you have to use TransactionTestCase.