为什么在 pytest-django 中使用 ThreadPoolExecutor 时会得到空的 django 查询集？

Question

我一直在尝试追踪一些并发代码中的一些错误，并想编写一个并行运行函数的测试。我使用 Django 和 postgres 作为我的数据库，并使用 pytest 和 pytest-django 进行测试。

为了运行我的函数，我使用ThreadPoolExecutor并简单地查询我的数据库并返回对象计数。这是 django shell 中的测试按预期工作：

>>> from concurrent.futures import *
>>> def count_accounts():
...     return Account.objects.all().count()
...
>>> count_accounts()
2
>>> with ThreadPoolExecutor(max_workers=1) as e:
...     future = e.submit(count_accounts)
...
>>> for f in as_completed([future]):
...     print(f.result())
...
2

但是，当我在 pytest 下运行此测试时，线程中的函数似乎返回空查询集：

class TestCountAccounts(TestCase):
    def test_count_accounts(self):
        def count_accounts():
            return Account.objects.all().count()

        initial_result = count_accounts()  # 2
        with ThreadPoolExecutor(max_workers=1) as e:
            future = e.submit(count_accounts)

        for f in as_completed([future]):
            assert f.result() == initial_result  # 0 != 2

无论如何，我可以在线程内进行调用以返回正确的值/正确访问数据库吗？

Answer 1

尝试使用TransactionTestCase而不是TestCase. TestCase用 包装类，atomic()并用 包装每个测试atomic()，因此线程很可能在创建测试数据的事务之外执行。

有关两者之间差异的更多信息：http://rahmonov.me/posts/testcase-vs-transactiontestcase-in-django/