As part of the work on CX-Metrics, a system for automatically recognizing demographic characteristics and customer behavior, by combining Artificial Intelligence algorithms and existing camera infrastructure (CCTV), we had to perform a number of tests, including tests on asynchronous-style code.
One of the examples is our proprietary server, whose task is to stream video directly to the frontend, so that on the one hand you can view the image from the cameras in a web browser, and on the other hand, keep the latency as low as possible (understood as the delay between the displayed film frames and the actual events having place in a store or shopping mall).
To write the server, we chose Python with the aiohttp package, which is based on the asynchronous programming paradigm, while the unit tests of this component were based on the Pytest framework. In this article, we want to show you how we approached testing async functions with the Pytest package, but we will precede this with a short introduction to asynchronous programming in Python – if you want to know more details, we recommend this article, which covers this topic much more.
Concurrent programming in Python
Concurrent programming in Python is a complicated topic due to the limitations of the interpreter (Global Interpreter Lock, GIL). Despite this, Python includes several packages in its standard library that allow for concurrent programming: multiprocessing – code that runs in separate processes that runs truly in parallel threading – code that runs in threads, due to the aforementioned GIL, threads are only concurrent asyncio – implements the asynchronous programming paradigm, which we will describe below.
Which solution we will use depends largely on our problem, in the case of the above-mentioned server, intermediating between the CCTV camera and the web browser, the use of asyncio turned out to be optimal.
Asyncio package
What may seem counterintuitive is the fact that async all operations are performed on a single thread. Thanks to this, synchronization between tasks is much easier than when using threading, while tasks share memory (as opposed to multiprocessing), which facilitates communication between them. The asyncio package introduces two new keywords in Python: async – precedes the definition of an async function, also called coroutine await – precedes an async function call, can only be used in a function marked async. event loop. When an await keyword is encountered, event loop can decide to interrupt the currently running task and allocate CPU time to another task. This approach is perfect for spending a significant amount of time waiting for I / O operations (I/O bound task), i.e. connection with a database, network queries, or receiving a video stream from an industrial camera … This solution is gaining popularity very quickly and more and more libraries are compatible with this programming paradigm (np. aiohttp, websockets, aiofiles, databases, read more here).
Below is an example of code written with asyncio: We can see that the defined functions are indeed executed concurrently by running the following command: Sample program printout:
By the way, it is worth noting that asynchronous functions can be called in traditional Python code by referring directly to the event loop.
Testing
As the number of users grows, so does the demand for testing the code written in this way. Despite the differences in the syntax of async functions, integration of asyncio with the Pytest framework requires only minimal changes. We can see it best on the example of tests for the long_computation function defined above. The first naive solution that comes to mind might be to add the async keyword to the test definition, as in the example below.
However, when you do this test, you will find that the question missed it, because this framework does not support asynchronous functions by default. To solve this problem, we can refer directly to the event loop.
Using the facilities available in the Pytest (@ Pytest.fixture), we create an event loop and guarantee the correct release of resources after the test is completed. We run our asynchronous function with the run_until_complete method. By running this test you will see that it does indeed work properly. This solution provides more customization (e.g. returning your own event_loop implementation), but we can also find a simpler way.
Pytest-asyncio pack
After installing the Pytest-asyncio plug-in, we have the @ Pytest.mark.asyncio decorator at our disposal, which causes the test to be treated as an asynchronous function. We can then use the await keyword, which increases the readability of the code.
Mocking
The second element that requires some modification when testing asynchronous functions is mocking (we will not delve into terminology and differences between stubs and mocks here, please refer to here). Mocking allows you to isolate the behavior of the tested component without testing the correctness of its dependencies. By default, the Mock class available in the unittest package does not support the asyncio syntax, but by inheriting from this class, we can write our own async / await-compatible implementation. Our overloaded call operator should return an async function.
Context manager patch mocks our long_computation function, so it won’t actually be called as a dependency of launch_tasks. In our example, an additional advantage of this solution is faster test execution.
Asynctest package
As was the case with the decorator we use, we don’t need to inherit from the Mock class. By installing the asynctest package, we have access to the CoroutineMock class, which performs exactly the same task.
Summary
As part of this article, we discussed the topic of testing Python code using the asyncio package and showed that adapting the Pytest framework in this direction does not require a lot of work from us. When testing asynchronous code, we must remember that it can only be executed after allocating resources through the event loop. Pytest thanks to its architecture allows us to easily add support for asyncio. Additionally, the Pytest-asyncio and asynctest packages provide us with the necessary tools, which makes adapting tests to async functions easier.
