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 异常可以使事情顺利进行。

所以我猜测试运行器在执行过程中被阻止了。

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