mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-07-24 03:35:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

[3.11] gh-99357: Close the event loop when it is no longer used in test_uncancel_structured_blocks (GH-99414) (#99424)

Browse source 
(cherry picked from commit 99972dc745)

Co-authored-by: Xiao Chen <chenxiao_7@163.com>
This commit is contained in:
Miss Islington (bot) 2022-11-14 10:52:10 -08:00 • committed by GitHub
parent 5971a65d07
commit 90de4b76b9
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
1 changed files with 23 additions and 20 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -639,27 +639,30 @@ class BaseTaskTests:
await asyncio.sleep(0)
return timed_out, structured_block_finished, outer_code_reached
# Test which timed out.
t1 = self.new_task(loop, make_request_with_timeout(sleep=10.0, timeout=0.1))
timed_out, structured_block_finished, outer_code_reached = (
loop.run_until_complete(t1)
)
self.assertTrue(timed_out)
self.assertFalse(structured_block_finished) # it was cancelled
self.assertTrue(outer_code_reached) # task got uncancelled after leaving
# the structured block and continued until
# completion
self.assertEqual(t1.cancelling(), 0) # no pending cancellation of the outer task
try:
# Test which timed out.
t1 = self.new_task(loop, make_request_with_timeout(sleep=10.0, timeout=0.1))
timed_out, structured_block_finished, outer_code_reached = (
loop.run_until_complete(t1)
)
self.assertTrue(timed_out)
self.assertFalse(structured_block_finished) # it was cancelled
self.assertTrue(outer_code_reached) # task got uncancelled after leaving
# the structured block and continued until
# completion
self.assertEqual(t1.cancelling(), 0) # no pending cancellation of the outer task
# Test which did not time out.
t2 = self.new_task(loop, make_request_with_timeout(sleep=0, timeout=10.0))
timed_out, structured_block_finished, outer_code_reached = (
loop.run_until_complete(t2)
)
self.assertFalse(timed_out)
self.assertTrue(structured_block_finished)
self.assertTrue(outer_code_reached)
self.assertEqual(t2.cancelling(), 0)
# Test which did not time out.
t2 = self.new_task(loop, make_request_with_timeout(sleep=0, timeout=10.0))
timed_out, structured_block_finished, outer_code_reached = (
loop.run_until_complete(t2)
)
self.assertFalse(timed_out)
self.assertTrue(structured_block_finished)
self.assertTrue(outer_code_reached)
self.assertEqual(t2.cancelling(), 0)
finally:
loop.close()
def test_cancel(self):