Mocking Asyncio Subprocess in Python with pytest

What this article is and is NOT

• It is NOT an introduction to mocking
• It is NOT a “copy-paste” project
• It IS a story and example of how to take existing examples/documentation and extend them to less common use-cases.

Background

Mocking in Python is a far easier task than I ever thought it would be. At the beginning of this week, I had never mocked a single function — it’s really that simple to pick up.

That said, some things are far easier to mock than others and some are far better documented. In particular, one area which I felt to be lacking was the mocking/patching of “objects” within a mock.

A prime example of this is using subprocess — if piping the result and using the .communicate() approach, simple mocking following the documentation will not be sufficient. Early on in my struggle, I stumbled upon this fantastic StackOverflow answer which gave a brilliant overview of the key concepts. This was sufficient for a while until I began to port my code over to use asyncio. Suddenly I was dealing with futures and numerous other new concepts and on top of all that, I was still trying to mock my results to avoid calling real file system commands.

After a few hours of trial and error, this is the solution I came to — it is by no means perfect and may not follow many best practices, but it most certainly works. I welcome any and all feedback on how this approach could be improved.

The Code

Here is an overview of the overall project structure:

├── Pipfile
├── Pipfile.lock
├── src
│   ├── __init__.py
│   └── utils
│       ├── AsyncioExample.py
│       └── __init__.py
└── test
    └── utils
        ├── TestAsyncioExample.py
        ├── __init__.py
        └── snapshots
            ├── __init__.py
            └── snap_TestAsyncioExample.py

I like to the Pipenv for all projects — I highly recommend it to keep your Python libraries separate and non-conflicting

Pipfile

For this basic example, let’s assume I want to perform an async-await ls command (obviously this was not my use case as there are far better ways to list files natively in Python)

AsyncioExample.py

This is very similar to the standard subprocess.Popen syntax except everything is awaited.

To test this file, we need some additional libraries (see the Pipfile above). Here is what a basic test would look like.

TestAsyncioExample.py

Let’s take a deeper dive here:

• Line 9: @asynctest... tells pytest we are patch the asyncio library (and specifically the create_subprocess_shell method). In a basic example, this is all that would be needed but our code uses the .communicate() “attribute” so a little more setup is required
• Line 12 sets up the mock that we will modify. This is a second mock which will be injected into the return value of mock_async_subproc (the parameter created by the line 9 annotation)
• Lines 13–21: Here is where I lost a lot of time — the response that .communicate() returns under the asyncio implementation is a Future (because it is awaited on line 15 of AsyncioExample.py )
• Lines 22 and 23 look familiar from the SO post above, the attributes are set and configured inside the secondary mock
• Line 24 finally sets the return value of our top-level mock, asyncio.create_subprocess_shell to be our second mock, process_mock

And that’s it — that’s all the setup required to mock an asyncio call to subprocess but the concept can be applied to any asyncio method, especially those which require additional “attributes” to be set.

(For those not familiar with snapshots, line 27 saves the output to a file [below] which can be verified and checked into a git repo for future test invocations to validate against).

snap_TestAsyncioExample.py

Wrapping Up

I hope this whirlwind tour of a less common mocking pattern can save some of you a lot of headache and Googling!

Update (2020–06–20)

I found myself reusing this logic across a number of different test classes so I refactored the logic above into a simple mock class which can easily be imported and used: