postgresql 使用ThreadPoolExecutor时在Django测试中保存到数据库

zpjtge22  于 11个月前  发布在  PostgreSQL
关注(0)|答案(1)|浏览(103)

我在尝试使用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来运行测试。任何关于为什么这种方法可能是错误的帮助或想法都很感激。

nzk0hqpo

nzk0hqpo1#

这对我很有效。把你的测试放在这个结构中:

from django.test import TransactionTestCase

class AsyncTestCase(TransactionTestCase):

    def test_my_code(self):
        ...

字符集
这会将数据保存在内存中,而不会提交到数据库。我猜线程代码在测试期间使用了不同的数据库连接。

相关问题