我一直在尝试追踪一些并发代码中的一些错误,并想编写一个并行运行函数的测试。我使用 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
我是否可以在线程内进行调用以返回正确的值/正确访问数据库?
最佳答案
尝试使用TransactionTestCase
而不是TestCase
。 TestCase
使用 atomic()
包装类,并使用 atomic()
包装每个测试,因此线程很可能在事务之外执行,其中正在创建您的测试数据。
有关两者之间差异的更多信息:http://rahmonov.me/posts/testcase-vs-transactiontestcase-in-django/
关于python - 为什么在 pytest-django 中使用 ThreadPoolExecutor 时会得到空的 django 查询集?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/56879960/