Add pytest fixture to clean up the IO loop #471
Description
I recently came across this error in my unit tests:
RuntimeError: This event loop is already running
The error doesn't come from streamz, it's from jupyter_core, in this recent change.
But it is triggered because my streamz tests don't clean up properly after them selves.
They leave the global async event loop, asyncio.get_event_loop(), in a running state.
To fix it I had to add this fixture to my unit tests:
@pytest.fixture(autouse=True)
def loop_cleaner():
yield None
close_io_loop()
def close_io_loop():
IOLoop().current().stop()
IOLoop().current().close()
assert not asyncio.get_event_loop().is_running()
asyncio.get_event_loop().close()
assert asyncio.get_event_loop().is_closed()
asyncio.set_event_loop(None)
It's a bit similar to the existing pristine_loop, but doesn't exactly do the same thing (afaict).
I hope this can help any one having similar issues.