我在尝试使用Django的TestCase来测试一段与ThreadPoolExecutor并行的代码时遇到了一个问题。在服务器上,一切都运行得很好(至少到目前为止没有已知的问题“手工”测试),但是当我运行我的测试时,它们的执行只是卡在保存()方法上。所以,基本上我有这样的东西:
class A(models.Model):
# something here
def start(self):
objectsB = []
for i in range(n):
objectB = B(i)
objectsB.append(objectB)
results = []
with concurrent.futures.ThreadPoolExecutor(max_workers=len(objectsB)) as executor:
futures = [executor.submit(objectB.start) for objectB in objectsB]
for future in concurrent.futures.as_completed(futures):
results.append(future.result())
return results
class B(models.Model):
# something here
def start(self):
# some logic here (including reading from database)
response = requests.post('http://www.externalapi.com')
# some logic here
print('during test this is printed')
self.save()
print('during test this is never reached')
字符集
当我在没有并发的情况下使用相同的代码时,只是顺序地执行objectB.start
,测试工作得很好。在objectsB的方法start
中阅读对数据库的调用似乎也可以工作。我正在使用Django 1.9,python 3.5,postgresql 9.5.3和nose来运行测试。任何关于为什么这种方法可能是错误的帮助或想法都很感激。
1条答案
按热度按时间nzk0hqpo1#
这对我很有效。把你的测试放在这个结构中:
字符集
这会将数据保存在内存中,而不会提交到数据库。我猜线程代码在测试期间使用了不同的数据库连接。