Тестирование, если исключение останавливает асинхронный цикл - PullRequest
Новая
       29

Тестирование, если исключение останавливает асинхронный цикл

0 голосов
/ 23 мая 2019

Я пишу тесты для кода Python, который запустит сервер и продюсера, используя AIOHTTP. Хотя сервер должен работать в фоновом режиме, этот производитель будет выполнять некоторые асинхронные задачи при инициализации сервера. Я использую Asyncio, но я новичок в этом. Я хочу обеспечить постепенное отключение, как указано здесь . Я хочу проверить, что при получении 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, и я высмеиваю все асинхронные функции с помощью asynctest.Mock () и сопрограммы set_producer с asynctest.CoroutineMock(). А при вызове ensure_future будет вызываться KeyboardInterrupt как побочный эффект, входящий в обработку исключений при вызове main.main (). Однако этот тест не пройден, поскольку исключение не возникло: 

Failed: DID NOT RAISE <class 'KeyboardInterrupt'>

Я также попытался упростить тест без with pytest.raises(KeyboardInterrupt) as excinfo:, и в этом случае тест завершился неудачно: 

concurrent.futures._base.CancelledError

Действительно ли мой тест неверен, или мне нужно изменить способ обработки исключения?

Заранее спасибо

...