Editor's note: This post was originally published in May, 2009 and was updated in December, 2024 to incorporate changes in Django and improvements suggested by our readers. It has also been tested for compatibility as of the Django 4.2 release.
At Caktus, we rely heavily on automated testing for web app development. We create tests for all the code we write, ideally before the code is written. We create tests for every bug we find and, resources permitting, ramp up the test suite with lots of random input and boundary testing.
Debugging concurrency issues or race conditions is challenging at best. Unit tests typically run only one at a time, or if they are run in parallel, precautions are taken to ensure they do not negatively impact each other. Even testing for concurrency bugs manually can be difficult if not impossible. There are only so many times you can double click the link in your web app that is generating some bizarre failure.
Using the Django test client, we created a little decorator that you can use in your unit tests to make sure a view doesn't blow up when it's called multiple times with the same arguments. If it does blow up, and you happen to be using PostgreSQL, chances are you can fix the issues by using Colin's previously posted require_lock decorator.
Below is the decorator for testing concurrency. You might find this useful, for example, to check for bugs in Django views that do a SELECT and then a subsequent INSERT or UPDATE non-atomically (or without checking that the data has not changed since the SELECT).
import threading
def test_concurrently(times):
"""
Add this decorator to small pieces of code that you want to test
concurrently to make sure they don't raise exceptions when run at the
same time. E.g., some Django views that do a SELECT and then a subsequent
INSERT might fail when the INSERT assumes that the data has not changed
since the SELECT.
"""
def test_concurrently_decorator(test_func):
def wrapper(*args, **kwargs):
exceptions = []
def call_test_func():
try:
test_func(*args, **kwargs)
except Exception as e:
exceptions.append(e)
raise
threads = []
for i in range(times):
threads.append(threading.Thread(target=call_test_func))
for t in threads:
t.start()
for t in threads:
t.join()
if exceptions:
raise Exception(f"test_concurrently intercepted {len(exceptions)} exceptions: {exceptions}")
return wrapper
return test_concurrently_decorator
To use this in a test, create a small function that includes the thread-safe code inside your test. Apply the decorator, passing the number of times you want to run the code simultaneously, and then call the function:
class MyTestCase(TestCase):
def test_registration_threaded(self):
url = reverse('toggle_registration')
@test_concurrently(15)
def toggle_registration():
# perform the code you want to test here; it must be thread-safe
# (e.g., each thread must have its own Django test client)
c = Client()
c.login(username='user@example.com', password='abc123')
response = c.get(url)
toggle_registration()
As with other unit tests, it is helpful to keep (or temporarily re-introduce) the original bug in the code so that you can verify that the test catches it. In particular, it may be useful to adjust the number of concurrent threads to the point where the test can reproduce the bug consistently. After that, you can fix the bug and verify that the test passes.
When running tests in different threads, each thread obtains its own database connection. Due to the internals of how Django runs tests using transactions (whereby the changes are never committed to the database), you may therefore receive errors where/in which the objects you created in one thread do not appear to exist in the other thread. For instance, in our earlier example, if some database changes are made within the test_registration_threaded method but before the toggle_registration function, those changes may not be visible within the toggle_registration when it's being run in a thread. Therefore, if the code being tested involves database queries, it may be necessary to use a TransactionTestCase subclass so that all threads can share access to the same data in the database:
from django.db import connection
class MyTestCase(TransactionTestCase):
serialized_rollback = True
def test_registration_threaded(self):
url = reverse('toggle_registration')
@test_concurrently(15)
def toggle_registration():
# perform the code you want to test here; it must be thread-safe
# (e.g., each thread must have its own Django test client)
c = Client()
c.login(username='user@example.com', password='abc123')
response = c.get(url)
# Prevent database connections being left open by the threads
connection.close()
toggle_registration()
This is how we handle testing Django views concurrently in 2024. If you have a better way, we'd love to hear about it.
Please share your thoughts in the comments below.
