Я пытался отследить некоторые ошибки в некотором параллельном коде и хотел написать тест, который выполнял функцию параллельно.Я использую Django с postgres в качестве базы данных и тестирую с использованием pytest и pytest-django.
Для запуска своей функции я использую ThreadPoolExecutor
и просто запрашиваю базу данных и возвращаю количество объектов.Вот тест в оболочке django, работающий как ожидалось:
>>> 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
Могу ли я в любом случае получить вызов внутри потока, чтобы вернуть правильное значение / получить доступ к БД правильно?