Created On Feb 27, 2025 With Tags python, asyncio, testing
TL;DR: To conquer the concurrency problem in pytest, multiple solutions has been explored. Starting with pytest-xdist, then a combination of pytest-asyncio with pytest-subtests, and finally built my own plugin, pytest-asyncio-concurrent, to overcome limitations.
def test_my_system(): send_event_to_my_system() for i in range(10): if verify_my_system_behavior(): return else: time.sleep(1) raise Exception("my system did not behave as expected")
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0collected 1 itemtests/my_test.py::test_my_system PASSED [100%]========================== 1 passed in 5.01s ==========================
This simple test case exemplifies the essence of many system tests. Send an event, verify the result periodically, and have a timeout. The example test case takes at most 10 seconds to finish, but in the real world, this can be considerably longer.
Life is good so far, until more test cases need to be covered.
@pytest.mark.parametrize("param", [...])def test_my_system(param): send_event_to_my_system(param) for i in range(10): if verify_my_system_behavior(param): return else: time.sleep(1) raise Exception("my system did not behave as expected")
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0collected 5 itemtests/my_test.py::test_my_system[1] PASSED [20%]tests/my_test.py::test_my_system[2] PASSED [40%]tests/my_test.py::test_my_system[3] PASSED [60%]tests/my_test.py::test_my_system[4] PASSED [80%]tests/my_test.py::test_my_system[5] PASSED [100%]========================== 5 passed in 25.03s ==========================
The tests still passing. However, it started to become a pain as it is time-consuming to develop, debug, and run them. The tests now take minutes to complete. As more and more tests were added, the situation worsened.
pytest-xdist provides a general method of bringing concurrency into pytest framework by executing tests on multiple processes.
$ python -m pytest -n auto========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.016 workers [5 items]scheduling tests via LoadScheduling[gw3] [ 20%] PASSED tests/test.py::test_my_system[4][gw2] [ 40%] PASSED tests/test.py::test_my_system[3][gw0] [ 60%] PASSED tests/test.py::test_my_system[1][gw4] [ 80%] PASSED tests/test.py::test_my_system[5][gw1] [100%] PASSED tests/test.py::test_my_system[2]========================== 5 passed in 5.87s ==========================
The -n CLI parameter specifies the number of processes to use for test execution. With -n auto, pytest-xdist will create a number of processes equal to the number of physical CPU cores.
This solution works effectively. As the test cases kept growing, the total number of tests exceeded the number of CPU cores pretty soon. In order to achieve the shortest test execution time (Yes, I am greedy.), I switched to -n ##, and kept increasing the number.
The number after -n kept increasing until our fragile dev server crashed due to too many worker processes. However, most of the time, those workers are simply sleep-ing, and occupying resources. Why should we spawn dozens of workers in such a scenario?
Let's use async to allow resource sharing across different test cases within single process. Although async has been around in Python since 3.4, pytest does not natively support asynchronous test cases.
pytest-asyncio bridges the gap of async and pytest, making async function test-able.
@pytest.mark.asyncio@pytest.mark.parametrize("param", [...])async def test_my_system(param): send_event_to_my_system(param) for i in range(10): if verify_my_system_behavior(param): return else: await asyncio.sleep(1) raise Exception("my system did not behave as expected")
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0plugins: asyncio-0.25.3asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=Nonecollected 5 itemtests/my_test.py::test_my_system[1] PASSED [20%]tests/my_test.py::test_my_system[2] PASSED [40%]tests/my_test.py::test_my_system[3] PASSED [60%]tests/my_test.py::test_my_system[4] PASSED [80%]tests/my_test.py::test_my_system[5] PASSED [100%]========================== 5 passed in 25.04s ==========================
Without thoroughly reviewing the documentation of pytest-asyncio, the tests took minutes again, async tests ran as if they were synchronous tests. We found ourselves back to the beginning.
It turns out pytest-asyncio wraps all async test functions as synchronous functions, making them consumable for pytest as regular functions.
Although pytest-asyncio does not allow tests to be run concurrently, it do introduce concurrency inside the scope of single test.
@pytest.mark.asyncioasync def test_my_system(): async def test_my_system_single(param): try: send_event_to_my_system(param) for i in range(10): if verify_my_system_behavior(param): return else: await asyncio.sleep(1) raise Exception("my system did not behave as expected") except Exception as e: logger.error(f"test case test_my_system_single{param} fail", exc_info=e) tasks = [test_my_system_single(param) for param in [...]] await asyncio.gather(*tasks)
We can incorporate different test cases into one test and execute them in a single loop. However, the downside is obvious and huge, all the error handling and dependency management become our responsibility, and these test cases disappear from test report. We pretty much lose the benefit of using a testing framework.
Basically, we were creating a bunch of subtests within one test case and organize them by ourselves. And fortunately, pytest-subtests can help us manage them in a more structured manner.
@pytest.mark.asyncioasync def test_my_system(subtests): async def test_my_system_single(param): with subtests.test(msg=f'test_my_system[{param}]'): send_event_to_my_system(param) for i in range(10): if verify_my_system_behavior(param): return else: await asyncio.sleep(1) tasks = [test_my_system_single(param) for param in [...]] await asyncio.gather(*tasks)
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0plugins: asyncio-0.25.3, subtests-0.14.1asyncio: mode=Mode.STRICT, asyncio_default_fixture_loop_scope=Nonecollected 1 itemtests/test.py::test_my_system [test_my_system[1]] SUBPASS [100%]tests/test.py::test_my_system [test_my_system[2]] SUBPASS [100%]tests/test.py::test_my_system [test_my_system[3]] SUBPASS [100%]tests/test.py::test_my_system [test_my_system[4]] SUBPASS [100%]tests/test.py::test_my_system [test_my_system[5]] SUBPASS [100%]tests/test.py::test_my_system PASSED [100%]================= 1 passed, 5 subtests passed in 5.01s ================
This approach is becoming solid. Tests run concurrently within the same process, and we retain most of the benefits from the framework, while the cost is some boilerplate required for each parent test.
The solution described above functions effectively, but is still suboptimal. Subtests are not first-class citizens in pytest, and the boilerplate is required in every parent test function. To minimize the boilerplate and eliminate the appearance of subtests, I built a plugin to bridge the gap.
pytest-asyncio-concurrent is designed to enable async tests to run concurrently while providing granular concurrency control.
@pytest.mark.asyncio_concurrent(group="my_system")@pytest.mark.parametrize("param", [...])async def test_my_system(param): send_event_to_my_system(param) for i in range(10): if verify_my_system_behavior(param): return else: await asyncio.sleep(1) raise Exception("my system did not behave as expected")
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0plugins: asyncio-concurrent-0.3.0collected 5 itemtests/my_test.py::test_my_system[1] PASSED [20%]tests/my_test.py::test_my_system[2] PASSED [40%]tests/my_test.py::test_my_system[3] PASSED [60%]tests/my_test.py::test_my_system[4] PASSED [80%]tests/my_test.py::test_my_system[5] PASSED [100%]========================== 5 passed in 5.04s ==========================
Using pytest-asyncio-concurrent is fairly straightforward, by marking tests with pytest.mark.asyncio_concurrent and assigning the same group name, those tests will run together. Conversely, tests marked with different group names will just go sequentially.
@pytest.mark.asyncio_concurrent()async def test_my_system_seperate(): await asyncio.sleep(2) assert verify()@pytest.mark.asyncio_concurrent(group="my_group")async def test_my_system_grouped_1(): await asyncio.sleep(1) assert verify()@pytest.mark.asyncio_concurrent(group="my_group")async def test_my_system_grouped_2(): await asyncio.sleep(1) assert verify()
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0plugins: asyncio-concurrent-0.3.0collected 3 itemtests/my_test.py::test_my_system_seperate PASSED [33%]tests/my_test.py::test_my_system_grouped_2 PASSED [67%]tests/my_test.py::test_my_system_grouped_1 PASSED [100%]========================== 3 passed in 3.03s ==========================
As mentioned above the plugin evolved from the solution pytest-asyncio + pytest-subtests, but the boilerplate has been incorporated into the framework. Thus async tests are collected and grouped during the test collection phase, and executed group by group in the execution phase.
I don't want to dive too deep into the design and implementation of pytest. So, long story short, Pytest maintains a SetupState to keep track of a stack of active nodes, beginning with Session, and ending with a Function, each node should be child of the previous node, in other words, we can not push multiple Functions onto the stack directly.
Therefore, one more layer of node, Group, has been created. Instead of pushing Functions onto the stack, the Group resides on the stack and manages async test functions, just like our approach in the pytest-subtests solution.
It's not easy to fully understand the latest state of the system during system tests execution. As a result, falling into infinite loops is not uncommon. A Timeout feature has been introduced for convenience.
@pytest.mark.asyncio_concurrent(group="my_group", timeout=10)async def test_infinite_loop(): while True: await asyncio.sleep(2)@pytest.mark.asyncio_concurrent(group="my_group", timeout=10)async def test_system(): await asyncio.sleep(1) assert verify()
========================= test session starts =========================platform linux -- Python 3.12.5, pytest-8.3.4, pluggy-1.5.0plugins: asyncio-concurrent-0.3.0collected 2 itemtests/my_test.py::test_infinite_loop FAILED [50%]tests/my_test.py::test_system PASSED [100%]============================== FAILURES ===============================_________________________ test_infinite_loop __________________________ @pytest.mark.asyncio_concurrent(group="my_group", timeout=10) async def test_infinite_loop(): while True:> await asyncio.sleep(2)...==================== 1 failed, 1 passed in 10.04s =====================
Note: This section addresses implementation details. Feel free to skip if you are not interested.
Going back to the solution pytest-asyncio + pytest-subtests, all tests executed as subtests are more or less sharing the same set of fixtures. This means we have to either manage the fixture lifecycle inside the test function or just literally let them share the same fixtures.
Moving to the plugin pytest-asyncio-concurrent, We are facing the same problem. Pytest registers each FixtureDef as a singleton across the session. The FixtureDef instance contains the metadata of the fixture, and is also in charge of storing its value once it is entered and clearing once exited.
The solution, while a bit hacky, is straightforward. We clone the FixtureDef instance before a function fixture is requested, and we maintain a cache of FixtureDef instances using Function as key.
When working with system tests, the testing strategy should reflect the model of the system under testing, particularly regarding concurrency.
pytest-xdist first if:
async.pytest-asyncio + pytest-subtests if :
async.I built pytest-asyncio-concurrent to bridge the gap after experimenting with different approaches. Welcome to try out pytest-asyncio-concurrent, and welcome any feedback!