asyncio testing inside notebooks

Recently I started to work on a project mixing lots of network IO and concurrent tasks. Due to these requirements asyncio seemed like a prefect fit. However this project was also my first big foray into the async world of python and I needed a bit of time to learn the ropes. As mentioned in a previous post, I do love to familiarize myself with new technology by experimenting in a notebook and only then to refactor the result into python packages.

In this blog post, I would like to show how to combine asyncio and pytest inside notebooks. IPython has had an excellent async support for quite some while now. However, running async tests inside IPython requires a bit of care. Below I describe what changes in ipytest were necessary.

But first, I would like to explore a bit how IPython's async support helps you to directly call async code in the notebook. For example, the following piece of code will directly work with IPython>=7.0.0.

import asyncio
await asyncio.sleep(0.1)

To support concurrent execution, asyncio uses a coordination layer called the event loop. At every await expression control is transferred back from the user's code to the event loop. It may then run a different asynchronous function or wait for external events. (For a more thorough introduction, see for example this great PyCon talk by Miguel Grinberg). Event loops are created per thread, i.e., each thread can execute its own collection of async functions. One detail that will become important is that IPython creates a default event loop for the main thread that is used to execute any top level async code.

The integration of asynchronous code into pytest is also pretty straightforward thanks to the excellent pytest-asyncio plugin. First install the packages via pip install pytest pytest-asyncio. Then, writing tests for asyncio code is as simple as:

import pytest 

@pytest.mark.asyncio
async def test_some_asyncio_code():
    actual = await do_something()
    assert actual == 42
    

async def do_something():
    await asyncio.sleep(0.1)
    return 42

However, things get tricky when trying to run these tests inside notebooks. The reason is that pytest-asyncio tries to creates its own event loop, which then conflicts with the main-thread event loop of IPython. Luckily, there is a simple solution: execute the tests in a separate thread. This thread will not have a pre-existing event loop and pytest-asyncio is free to create its own.

Since version 0.7.0, ipytest supports exactly this behavior by passing run_in_thread=True to config. First install the most recent version via pip install -U ipytest. Then configure ipytest:

# import ipytest
import ipytest

# expose the notebook name
__file__ = 'Post.ipynb'

ipytest.config(
    rewrite_asserts=True, addopts=['-qq'], 
    
    # run in a separate thread
    run_in_thread=True,
)

And finally run tests via:

ipytest.run()

And that's all that needs to be done to run asyncio tests inside jupyter notebooks. If you have any feedback or comments, feel free to reach out to me on twitter @c_prohm or post an issue on github.