問題描述:
使用django自帶的test做測試,嘗試去數據庫中取數據,主線程中沒有問題,非主線程中取不到數據。
示例代碼:
class MyTestCase(TestCase):
def setUp(self):
MyModel.objects.create(k='k0', v='v0')
MyModel.objects.create(k='k1', v='v1')
def test_multithread(self):
kv = MyModel.objects.get(k__exact='k0')
print kv
def func():
kv = MyModel.objects.get(k__exact='k0')
print kv
t = threading.Thread(target=func, args=())
t.setDaemon(False)
t.start()
主線程中的查詢語句正確輸出,func函數中的查詢語句報錯(DoesNotExist: MyModel matching query does not exist.)
解決:
1)不是多線程的原罪,只是數據庫的線程安全策略,鎖的時間比較長。嘗試把func函數改成這樣:
def func():
time.sleep(10)
kv = MyModel.objects.get(k__exact='k0')
print kv
能運行成功。所以一開始我猜測是鎖的時間比較長。。。
2)真正的問題是這樣的,MyTestCase繼承自TestCase,而在TestCase中數據庫的改動不會commit(或者不會立即commit),所以在另外一個
線程中看不到這些數據。Django提供的另一個測試基類(TransactionTestCase)可以解決這個問題,試試
class MyTestCase(TransactionTestCase)
就可以運行正確。
參考:http://stackoverflow.com/questions/10948537/database-errors-in-django-when-using-threading
補充:
如果以上方法仍然解決不了這個問題,可以選擇不在內存中建測試數據庫,在setting.py中加入:
import sys
if 'test' in sys.argv:
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': ':memory:',
'TEST':
{
'NAME': 'test_db.sqlite3',
}
},
}
和TransactionTestCase 一起使用。
