为什么在将 ThreadPoolExecutor 与 pytest-django 一起使用时会得到空的 django 查询集?
Why do I get empty django querysets when using ThreadPoolExecutor with pytest-django?
我一直在尝试追踪一些并发代码中的错误,并想编写一个 运行 并行函数的测试。我正在使用带有 postgres 的 Django 作为我的数据库,并使用 pytest 和 pytest-django 进行测试。
为了 运行 我的功能,我正在使用 ThreadPoolExecutor
并简单地查询我的数据库并 return 计算对象的数量。这是 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 下的测试时,它看起来像线程中的函数 returns 空查询集:
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
无论如何我可以在线程内调用 return 正确的 value/access 数据库吗?
尝试使用 TransactionTestCase
而不是 TestCase
。 TestCase
用 atomic()
包装 class,用 atomic()
包装每个测试,因此线程很可能在创建测试数据的事务之外执行。
有关两者区别的更多信息:http://rahmonov.me/posts/testcase-vs-transactiontestcase-in-django/
根据上面@marquee 的回答,如果您使用 pytest 和 django 并在其中一个测试中使用 futures
,则需要修改测试函数装饰器。而不仅仅是
@pytest.mark.django_db`
def test_something():
...
使用:
@pytest.mark.django_db(transaction=True)`
def test_something():
...
达到相同的结果。
我一直在尝试追踪一些并发代码中的错误,并想编写一个 运行 并行函数的测试。我正在使用带有 postgres 的 Django 作为我的数据库,并使用 pytest 和 pytest-django 进行测试。
为了 运行 我的功能,我正在使用 ThreadPoolExecutor
并简单地查询我的数据库并 return 计算对象的数量。这是 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 下的测试时,它看起来像线程中的函数 returns 空查询集:
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
无论如何我可以在线程内调用 return 正确的 value/access 数据库吗?
尝试使用 TransactionTestCase
而不是 TestCase
。 TestCase
用 atomic()
包装 class,用 atomic()
包装每个测试,因此线程很可能在创建测试数据的事务之外执行。
有关两者区别的更多信息:http://rahmonov.me/posts/testcase-vs-transactiontestcase-in-django/
根据上面@marquee 的回答,如果您使用 pytest 和 django 并在其中一个测试中使用 futures
,则需要修改测试函数装饰器。而不仅仅是
@pytest.mark.django_db`
def test_something():
...
使用:
@pytest.mark.django_db(transaction=True)`
def test_something():
...
达到相同的结果。