mirror of https://github.com/python/cpython
bpo-44911: Fixed IsolatedAsyncioTestCase from throwing an exception on leaked tasks (GH-27765)
This commit is contained in:
parent
ad0a8a9c62
commit
2cb1a6806c
|
@ -135,7 +135,7 @@ class IsolatedAsyncioTestCase(TestCase):
|
|||
task.cancel()
|
||||
|
||||
loop.run_until_complete(
|
||||
asyncio.gather(*to_cancel, loop=loop, return_exceptions=True))
|
||||
asyncio.gather(*to_cancel, return_exceptions=True))
|
||||
|
||||
for task in to_cancel:
|
||||
if task.cancelled():
|
||||
|
|
|
@ -216,6 +216,26 @@ class TestAsyncCase(unittest.TestCase):
|
|||
output = test.run()
|
||||
self.assertFalse(output.wasSuccessful())
|
||||
|
||||
def test_cancellation_hanging_tasks(self):
|
||||
cancelled = False
|
||||
class Test(unittest.IsolatedAsyncioTestCase):
|
||||
async def test_leaking_task(self):
|
||||
async def coro():
|
||||
nonlocal cancelled
|
||||
try:
|
||||
await asyncio.sleep(1)
|
||||
except asyncio.CancelledError:
|
||||
cancelled = True
|
||||
raise
|
||||
|
||||
# Leave this running in the background
|
||||
asyncio.create_task(coro())
|
||||
|
||||
test = Test("test_leaking_task")
|
||||
output = test.run()
|
||||
self.assertTrue(cancelled)
|
||||
|
||||
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
|
|
@ -0,0 +1 @@
|
|||
:class:`~unittest.IsolatedAsyncioTestCase` will no longer throw an exception while cancelling leaked tasks. Patch by Bar Harel.
|
Loading…
Reference in New Issue