summaryrefslogtreecommitdiffstats
path: root/Lib/test/test_asyncio/test_timeouts.py
diff options
context:
space:
mode:
authorMiss Islington (bot) <31488909+miss-islington@users.noreply.github.com>2022-07-26 10:16:12 (GMT)
committerGitHub <noreply@github.com>2022-07-26 10:16:12 (GMT)
commit19d953682e86ad51b19313ddba8ddb2602d6efd3 (patch)
tree27bce6f383b4627eeccdf45516ef66b9426c484a /Lib/test/test_asyncio/test_timeouts.py
parent2fb64a0687c36c22933bae38970daa3c9a1ee84f (diff)
downloadcpython-19d953682e86ad51b19313ddba8ddb2602d6efd3.zip
cpython-19d953682e86ad51b19313ddba8ddb2602d6efd3.tar.gz
cpython-19d953682e86ad51b19313ddba8ddb2602d6efd3.tar.bz2
gh-95051: ensure that timeouts scheduled with `asyncio.Timeout` that have already expired are deliverered promptly (GH-95109) (GH-95216)
Co-authored-by: Kumar Aditya <59607654+kumaraditya303@users.noreply.github.com> (cherry picked from commit 0c6f898005099be189ee65bcfda659f5fc13b802) Co-authored-by: Thomas Grainger <tagrain@gmail.com>
Diffstat (limited to 'Lib/test/test_asyncio/test_timeouts.py')
-rw-r--r--Lib/test/test_asyncio/test_timeouts.py24
1 files changed, 24 insertions, 0 deletions
diff --git a/Lib/test/test_asyncio/test_timeouts.py b/Lib/test/test_asyncio/test_timeouts.py
index ef1ab0a..3a54297 100644
--- a/Lib/test/test_asyncio/test_timeouts.py
+++ b/Lib/test/test_asyncio/test_timeouts.py
@@ -106,6 +106,30 @@ class TimeoutTests(unittest.IsolatedAsyncioTestCase):
self.assertLess(t1-t0, 2)
self.assertTrue(t0 <= cm.when() <= t1)
+ async def test_timeout_zero_sleep_zero(self):
+ loop = asyncio.get_running_loop()
+ t0 = loop.time()
+ with self.assertRaises(TimeoutError):
+ async with asyncio.timeout(0) as cm:
+ await asyncio.sleep(0)
+ t1 = loop.time()
+ self.assertTrue(cm.expired())
+ # 2 sec for slow CI boxes
+ self.assertLess(t1-t0, 2)
+ self.assertTrue(t0 <= cm.when() <= t1)
+
+ async def test_timeout_in_the_past_sleep_zero(self):
+ loop = asyncio.get_running_loop()
+ t0 = loop.time()
+ with self.assertRaises(TimeoutError):
+ async with asyncio.timeout(-11) as cm:
+ await asyncio.sleep(0)
+ t1 = loop.time()
+ self.assertTrue(cm.expired())
+ # 2 sec for slow CI boxes
+ self.assertLess(t1-t0, 2)
+ self.assertTrue(t0 >= cm.when() <= t1)
+
async def test_foreign_exception_passed(self):
with self.assertRaises(KeyError):
async with asyncio.timeout(0.01) as cm: