Here's how pytest-asyncio implements the event_loop fixture:
|
@pytest.fixture |
|
def event_loop(request): |
|
"""Create an instance of the default event loop for each test case.""" |
|
loop = asyncio.get_event_loop_policy().new_event_loop() |
|
yield loop |
|
loop.close() |
Note that loop.close() is the sole cleanup code.
Compare this with the implementation of asyncio.run():
loop = events.new_event_loop()
try:
events.set_event_loop(loop)
if debug is not None:
loop.set_debug(debug)
return loop.run_until_complete(main)
finally:
try:
_cancel_all_tasks(loop)
loop.run_until_complete(loop.shutdown_asyncgens())
loop.run_until_complete(loop.shutdown_default_executor())
finally:
events.set_event_loop(None)
loop.close()Before calling loop.close(), asyncio.run() does this:
- It attempts to cancel outstanding tasks with
_cancel_all_tasks().
- It calls
loop.shutdown_asyncgens()
- It calls
loop.shutdown_default_executor().
(I think (2) and (3) correspond to what the documentation calls "finalizing asynchronous generators and closing the threadpool.")
I only have a tenuous grasp on these asyncio internals, but it seems like the event_loop fixture should do this, too? And so should any other place in pytest-asyncio where the event loop is shut down, if there are more.
Here's how
pytest-asyncioimplements theevent_loopfixture:pytest-asyncio/pytest_asyncio/plugin.py
Lines 224 to 229 in 6ec7647
Note that
loop.close()is the sole cleanup code.Compare this with the implementation of
asyncio.run():Before calling
loop.close(),asyncio.run()does this:_cancel_all_tasks().loop.shutdown_asyncgens()loop.shutdown_default_executor().(I think (2) and (3) correspond to what the documentation calls "finalizing asynchronous generators and closing the threadpool.")
I only have a tenuous grasp on these
asynciointernals, but it seems like theevent_loopfixture should do this, too? And so should any other place inpytest-asynciowhere the event loop is shut down, if there are more.