Python unittest + asyncio висит навсегда

Question

Почему следующий тест висит навсегда?

import asyncio
import unittest


class TestCancellation(unittest.IsolatedAsyncioTestCase):

    async def test_works(self):
        task = asyncio.create_task(asyncio.sleep(5))
        await asyncio.sleep(2)
        task.cancel()
        await task


if __name__ == '__main__':
    unittest.main()

Answer 1

Перехват исключения CancelledError при ожидании отмененной задачи делает вещи go гладкими.

Так что я думаю, что бегун теста задерживается в действии.

import asyncio
import unittest


class TestCancellation(unittest.IsolatedAsyncioTestCase):

    async def test_works(self):
        task = asyncio.create_task(asyncio.sleep(5))
        await asyncio.sleep(2)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            print("Task Cancelled already")

if __name__ == '__main__':
    unittest.main()

производит

unittest-hang $ python3.8 test.py 
Task Cancelled already
.
----------------------------------------------------------------------
Ran 1 test in 2.009s

OK

Я игнорирую, нужно ли вам ждать отмененную задачу или нет.

Если вы должны, поскольку вы, похоже, полностью проверяете его отмену, то поймайте исключение.

Если нет, то просто избегайте этого, так как создание задачи запускает ее немедленно и не нужно ждать снова

import asyncio
import unittest


class TestCancellation(unittest.IsolatedAsyncioTestCase):

    async def test_works(self):
        task = asyncio.create_task(asyncio.sleep(5))
        await asyncio.sleep(2)
        task.cancel()
        # await task

if __name__ == '__main__':
    unittest.main()

производит

unittest-hang $ python3.8 test.py 
.
----------------------------------------------------------------------
Ran 1 test in 2.009s

OK

Answer 2

Как указано в комментарии @Pynchia, пример решения:

import asyncio
import unittest


class TestCancellation(unittest.IsolatedAsyncioTestCase):

    async def test_works(self):
        task = asyncio.create_task(asyncio.sleep(5))
        await asyncio.sleep(2)
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            print("main(): cancel_me is cancelled now")


if __name__ == '__main__':
    unittest.main()

Решение взято из документации asyncio.Task.cancel . Документация также объясняет это поведение:

Запрос отмены задачи.

Это позволяет исключить исключение CancelledError в завернутую сопрограмму в следующем цикле события l oop.

У сопрограммы есть шанс очистить или даже отклонить запрос, подавив исключение с помощью попытки ... ... кроме CancelledError ... finally блока. Следовательно, в отличие от Future.cancel (), Task.cancel () не гарантирует отмены Задачи, хотя полное подавление отмены не является распространенным явлением и активно не поощряется.