测试异常是否停止asyncio循环

Question

我正在为Python代码编写测试，该代码将使用AIOHTTP启动服务器和生产者。虽然服务器必须在后台运行，但此生产者将在初始化服务器时执行一些异步任务。 我正在使用Asyncio，但我对它还很陌生。 我想提供一个优雅的关闭，如here所示。我想测试一下，当接收到Ctrl-C（KeyboardInterrupt）时，异常处理将被有效地触发，以停止并关闭循环和任务。你知道吗

主要功能代码如下：

def main(foo):
    loop = asyncio.get_event_loop()
    prod = Producer()
    try:
        asyncio.ensure_future(webserver.start())
        asyncio.ensure_future(prod.set_producer())
        loop.run_forever()
    except Exception as e:
        logging.error("Error: %s", str(e))
    except KeyboardInterrupt:
        logger.info("Canceling pending tasks and stopping the loop")
        asyncio.gather(*asyncio.Task.all_tasks()).cancel()
        logger.info("Stopping the loop")
        loop.stop()
        logger.info("Received Ctrl+C")
        loop.close()

我要写的测试是：

@pytest.mark.asyncio
async def test_main_exception_correcly(event_loop):
    with pytest.raises(KeyboardInterrupt) as excinfo:
        event_loop.run_until_complete = asynctest.Mock()
        main.prod = asynctest.Mock()
        main.prod.set_producer = asynctest.CoroutineMock()
        main.asyncio.ensure_future = synctest.Mock(side_effect=KeyboardInterrupt())
        event_loop.close = asynctest.Mock()
        main.main()
        event_loop.close.assert_called()

我知道行with pytest.raises(KeyboardInterrupt) as excinfo将能够捕获任何KeyboardInterrupt，我用异步测试.模拟（）和set\u producer与asynctest.CoroutineMock()的协程。当调用ensure_future时，会产生一个KeyboardInterrupt作为副作用，在调用时进入异常处理梅因。梅因(). 但是，此测试失败，表示未引发异常：

Failed: DID NOT RAISE <class 'KeyboardInterrupt'>

我还尝试使测试更简单，没有with pytest.raises(KeyboardInterrupt) as excinfo:，在本例中，测试失败的原因是：

concurrent.futures._base.CancelledError

是我的测试真的错了，还是我需要改变处理异常的方式？你知道吗

提前谢谢