Stowage/cpython
2
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-11-10 18:42:04 +00:00
Code Issues Projects Releases Packages Wiki Activity

[3.14] gh-86802: Fix asyncio memory leak; shielded task exceptions log once through the exception handler (gh-134331) (gh-134343)

Browse source 
(cherry picked from commit f695eca60c)

Co-authored-by: Christian Harries <68507104+ChristianHrs@users.noreply.github.com>
Co-authored-by: Łukasz Langa <lukasz@langa.pl>
This commit is contained in:
Miss Islington (bot) 2025-05-20 19:43:09 +02:00 • committed by GitHub
parent 6ce2045139
commit efdaae5b35
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 72 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

40
Lib/test/test_asyncio/test_tasks.py
View file

@ -2116,6 +2116,46 @@ def test_shield_cancel_outer(self):
self.assertTrue(outer.cancelled())
self.assertEqual(0, 0 if outer._callbacks is None else len(outer._callbacks))
def test_shield_cancel_outer_result(self):
mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
inner = self.new_future(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
inner.set_result(1)
test_utils.run_briefly(self.loop)
mock_handler.assert_not_called()
def test_shield_cancel_outer_exception(self):
mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
inner = self.new_future(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
inner.set_exception(Exception('foo'))
test_utils.run_briefly(self.loop)
mock_handler.assert_called_once()
def test_shield_duplicate_log_once(self):
mock_handler = mock.Mock()
self.loop.set_exception_handler(mock_handler)
inner = self.new_future(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
outer = asyncio.shield(inner)
test_utils.run_briefly(self.loop)
outer.cancel()
test_utils.run_briefly(self.loop)
inner.set_exception(Exception('foo'))
test_utils.run_briefly(self.loop)
mock_handler.assert_called_once()
def test_shield_shortcut(self):
fut = self.new_future(self.loop)
fut.set_result(42)