CODEWITHSUNDEEP
pythonpython-3.xunit-testingpython-asynciopython-3.8
Andrei Moiseev
Andrei Moiseev

Reputation: 4074

Python unittest + asyncio hangs forever

Why does the following test hang forever?

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

Upvotes: 4

Views: 4079

Answers (2)

Pynchia
Pynchia

Reputation: 11580

Catching the CancelledError exception when awaiting the cancelled task makes things go smooth.

So I guess the test runner gets held up in the act.

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

produces

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

OK

I ignore whether you must await the cancelled task or not.

If you must, since you seem to be testing its cancellation fully, then catch the exception.

If not, then just avoid it, since creating a task starts it immediately and there is no need to await again

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

produces

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

OK

Upvotes: 2

fabianegli
fabianegli

Reputation: 2246

As by the comment from @Pynchia an example solution:

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

The solution is taken from the asyncio.Task.cancel documentation. The documentation also explains this behavior:

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.

Upvotes: 0

Related Questions