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

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 文档。文档还解释了这种行为：

Request the Task to be cancelled.

This arranges for a CancelledError exception to be thrown into the wrapped coroutine on the next cycle of the event loop.

The coroutine then has a chance to clean up or even deny the request by suppressing the exception with a try … … except CancelledError … finally block. Therefore, unlike Future.cancel(), Task.cancel() does not guarantee that the Task will be cancelled, although suppressing cancellation completely is not common and is actively discouraged.

Python unittest + asyncio 永远挂起

Python unittest + asyncio hangs forever

python

unit-testing

python-3.x

python-asyncio

python-3.8