Why do I get empty django querysets when using ThreadPoolExecutor with pytest-django?

Question

I have been trying to track down some bugs in some concurrent code and wanted to write a test that ran a function in parallel. I am using Django with postgres as my database and testing using pytest and pytest-django.

To run my function, I am using ThreadPoolExecutor and simply querying my database and returning a count of objects. Here is the test in the django shell working as expected:

>>> 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

However, when I run this as a test under pytest, it appears like the function in the thread returns empty querysets:

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

Is there anyway I can get the call inside the thread to return the correct value/access the db correctly?

Geometer

Answer 1

Try using TransactionTestCase instead of TestCase. TestCase wraps the class with atomic() and each test with atomic(), so it's likely that the thread is executing outside of the transaction where your test data is being created.

For more information on the difference between the two: http://rahmonov.me/posts/testcase-vs-transactiontestcase-in-django/

Klystron

Answer 2

Following on @marquee's answer above, if you are using pytest and django and using futures in one of your tests, you'll need to modify your test function decorator. Instead of just

@pytest.mark.django_db`
def test_something():
    ...

Use:

@pytest.mark.django_db(transaction=True)`
def test_something():
    ...

which achieves the same result.

Fleece