Skip to content
Prev Previous commit
Next Next commit
Remove tests for simultaneous *timeout* and *inner becoming done*.
The existing method of setting identical timeout / sleep lengths is not reliable.
  • Loading branch information
twisteroidambassador committed Oct 26, 2022
commit c834122db8f08928b2e555801314f877c7bfa5d2
35 changes: 13 additions & 22 deletions Lib/test/test_asyncio/test_waitfor.py
Original file line number Diff line number Diff line change
Expand Up @@ -295,18 +295,24 @@ async def simultaneous_self_cancel_and_inner_result(
waitfor_timeout,
inner_action,
):
"""Construct scenario where external cancellation, exceeding timeout and
awaitable becoming done all happen simultaneously.
"""Construct scenario where external cancellation and
awaitable becoming done happen simultaneously.
inner_acion is one of 'cancel', 'exception' or 'result'.
if do_cancel, the wait_for() call is cancelled at the same time when
inner_action is executed.
If waitfor_timeout == 0.1, then the timeout should happen at the same
time as well."""
Make sure waitfor_timeout > 0.1. Trying to make it == 0.1 is not
a reliable way to make timeout happen at the same time as the above.
"""
loop = asyncio.get_running_loop()
inner = loop.create_future()
waitfor_task = asyncio.create_task(
asyncio.wait_for(inner, timeout=waitfor_timeout))
await asyncio.sleep(0.1)
# Even if waitfor_timeout == 0.1, there's still no guarantee whether the
# timer handler (or similar) in wait_for() or code below in this
# coroutine executes first. If the timer handler executes first, then
# inner will be cancelled(), and code below will raise
# InvalidStateError.
if inner_action == 'cancel':
inner.cancel()
elif inner_action == 'exception':
Expand All @@ -319,41 +325,26 @@ async def simultaneous_self_cancel_and_inner_result(
return await waitfor_task

async def test_simultaneous_self_cancel_and_inner_result(self):
for timeout in (0.1, 10, None):
for timeout in (10, None):
with self.subTest(waitfor_timeout=timeout):
with self.assertRaises(asyncio.CancelledError):
await self.simultaneous_self_cancel_and_inner_result(
True, timeout, 'result')

async def test_simultaneous_self_cancel_and_inner_exc(self):
for timeout in (0.1, 10, None):
for timeout in (10, None):
with self.subTest(waitfor_timeout=timeout):
with self.assertRaises(RuntimeError):
await self.simultaneous_self_cancel_and_inner_result(
True, timeout, 'exception')

async def test_simultaneous_self_cancel_and_inner_cancel(self):
for timeout in (0.1, 10, None):
for timeout in (10, None):
with self.subTest(waitfor_timeout=timeout):
with self.assertRaises(asyncio.CancelledError):
await self.simultaneous_self_cancel_and_inner_result(
True, timeout, 'cancel')

async def test_simultaneous_inner_result_and_timeout(self):
result = await self.simultaneous_self_cancel_and_inner_result(
False, 0.1, 'result')
self.assertEqual(result, 'inner result')

async def test_simultaneous_inner_exc_and_timeout(self):
with self.assertRaises(RuntimeError):
await self.simultaneous_self_cancel_and_inner_result(
False, 0.1, 'exception')

async def test_simultaneous_inner_cancel_and_timeout(self):
with self.assertRaises(asyncio.CancelledError):
await self.simultaneous_self_cancel_and_inner_result(
False, 0.1, 'cancel')


if __name__ == '__main__':
unittest.main()