Given this code:
class ImportTest(TestCase):
account = None
def test_atomic(self):
def import_task():
print Account.objects.all()
threads = []
self.account = Account.objects.create(name='abc')
for i in range(10):
t = threading.Thread(target=import_task)
threads.append(t)
t.start()
for t in threads:
t.join()
The threads prints empty record set, but if I make it to extend TransactionTestCase
as below:
class ImportTest(TransactionTestCase):
account = None
def test_atomic(self):
def import_task():
print Account.objects.all()
threads = []
self.account = Account.objects.create(name='abc')
for i in range(10):
t = threading.Thread(target=import_task)
threads.append(t)
t.start()
for t in threads:
t.join()
This will print out the record created.
Could someone please explain this behaviour?
Because TestCase
runs inside of a transaction (this is an implementation detail designed to improve performance), you shouldn't use it for anything that is itself testing or relying on transactions. This is explained in the documentation.
What's happening in this case is probably database- and isolation-level-dependent, but my guess is that: the test is running inside of an open transaction because you're using TestCase
; that transaction remains open until the test is over and it's rolled back; the threads are creating their own connections to the database; and due to the isolation level they are not able to see the object created in the still-open main transaction.
The good news is that you've found the solution: use TransactionTestCase
. The test process will run in the regular autocommit mode, so the create()
commits to the database before the other threads do their lookups.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With