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 has long been a nightmare. 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, I 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.
Here's the decorator for testing concurrency:
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, 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('test_concurrently intercepted %s exceptions: %s' % (len(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 testRegistrationThreaded(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()
