Python unittest + asyncio永远挂起答案

【问题标题】：Python unittest + asyncio hangs foreverPython unittest + asyncio永远挂起
【发布时间】：2020-02-20 21:24:38
【问题描述】：

为什么下面的测试永远挂起？

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()

【问题讨论】：

如果您在等待取消的任务时发现CancelledError 异常，事情就会顺利进行。所以我猜测试运行器没有正确完成它的工作
您可能会获得更多见解here
我发现 pytest-asyncio 在异步测试方面比 unitest lib 更可靠。

标签： python python-3.x unit-testing python-asyncio python-3.8

【解决方案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

【讨论】：

这并不能解释为什么 OP 的示例永远挂起。等待取消的任务应该会泄漏一个CancelledError，这应该由 unittest 报告，但事实并非如此。
@a_guest 点确实被拿走了。我做了一些测试并修改了答案。目前尚不清楚 OP 是否需要等待取消的任务。作为一个测试，它对我来说没有多大意义，除非他正在测试测试运行器本身
@Pynchia 刚刚发现普通的raise asyncio.CancelledError() 也会挂起测试。可能是unittest 中的一个错误。

【解决方案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 异常被抛出到在事件循环的下一个循环中包装协程。

然后协程就有机会清理甚至拒绝请求通过尝试抑制异常…………除了 CancelledError…… 最后阻塞。因此，与 Future.cancel() 不同，Task.cancel() 确实不保证任务会被取消，虽然压制完全取消并不常见，我们强烈反对。

【讨论】：