Speed up Django transaction hooks tests

transaction.on_commit

django-transactions-hooks was merged to Django core in 1.9 release. It is a robust solution for performing actions only after your database commits, queuing tasks, sending emails, posting data to external services, …

transaction.on_commit(lambda: some_celery_task.delay('arg1'))

Avoid TransactionTestCase (SnailTestCase)

If your tests are based on TestCase no commit is ever done, you will be forced to adopt TransactionTestCase. Your tests execution time will raise dramatically. Remember what TransactionTestCase does:

  • resets DB after every test execution
  • fixtures are loaded for every test execution
  • Can’t use SetUpTestData to load shared data once per test class

It is a high price and it’s reasonable to have the temptation to leave your tests without checking that delayed code portion. Or maybe you are a bullheaded tester and have tattooed duck typing and mocking in your head.

“If it walks like a duck and it quacks like a duck, then it must be a duck.”

Dive into Django DB backend

Django DB backend is not tricky. We can identify where functions are saved for later execution. Have a look at django.db.backends.base.base.BaseDatabaseWrapper.on_commit. Postponed functions are stored in self.run_on_commit.

def on_commit(self, func):
if self.in_atomic_block:
# Transaction in progress; save for execution on commit.
self.run_on_commit.append((set(self.savepoint_ids), func))
elif not self.get_autocommit():
raise TransactionManagementError('on_commit() cannot be used in manual transaction management')
else:
# No transaction in progress and in autocommit mode; execute
# immediately.
func()

Also it not hard to find out where self.run_on_commit is iterated to execute delayed actions.

def run_and_clear_commit_hooks(self):
self.validate_no_atomic_block()
current_run_on_commit = self.run_on_commit
self.run_on_commit = []
while current_run_on_commit:
sids, func = current_run_on_commit.pop(0)
func()

Hack it

In this point we are capable to implement a solution forcing the execution of run_and_clear_commit_hooks() iterating through databases connections.

def run_commit_hooks(self):
"""
Fake transaction commit to run delayed on_commit functions
:return:
"""
for db_name in reversed(self._databases_names()):
with mock.patch('django.db.backends.base.base.BaseDatabaseWrapper.validate_no_atomic_block', lambda a: False):
transaction.get_connection(using=db_name).run_and_clear_commit_hooks()

Code is simple, just remember to mock validate_no_atomic_block to fake validation of active transaction.

Enjoy your fast tests

Just test transactions using TestCase and calling run_commit_hook whenever transaction hooks executions is required for testing.

class TestClass1(TestCase):
    def run_commit_hooks(self):
....
    def test(self):
# test code with on_commit delayed actions
....
# force delayed actions
self.run_commit_hooks()
# assert delayed actions
....

NOTE: Consider setUp might have postponed actions too as they are inside same transaction.