1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
|
import asyncio
import unittest
import time
from test import support
def tearDownModule():
asyncio.set_event_loop_policy(None)
# The following value can be used as a very small timeout:
# it passes check "timeout > 0", but has almost
# no effect on the test performance
_EPSILON = 0.0001
class SlowTask:
""" Task will run for this defined time, ignoring cancel requests """
TASK_TIMEOUT = 0.2
def __init__(self):
self.exited = False
async def run(self):
exitat = time.monotonic() + self.TASK_TIMEOUT
while True:
tosleep = exitat - time.monotonic()
if tosleep <= 0:
break
try:
await asyncio.sleep(tosleep)
except asyncio.CancelledError:
pass
self.exited = True
class AsyncioWaitForTest(unittest.IsolatedAsyncioTestCase):
async def test_asyncio_wait_for_cancelled(self):
t = SlowTask()
waitfortask = asyncio.create_task(
asyncio.wait_for(t.run(), t.TASK_TIMEOUT * 2))
await asyncio.sleep(0)
waitfortask.cancel()
await asyncio.wait({waitfortask})
self.assertTrue(t.exited)
async def test_asyncio_wait_for_timeout(self):
t = SlowTask()
try:
await asyncio.wait_for(t.run(), t.TASK_TIMEOUT / 2)
except asyncio.TimeoutError:
pass
self.assertTrue(t.exited)
async def test_wait_for_timeout_less_then_0_or_0_future_done(self):
loop = asyncio.get_running_loop()
fut = loop.create_future()
fut.set_result('done')
ret = await asyncio.wait_for(fut, 0)
self.assertEqual(ret, 'done')
self.assertTrue(fut.done())
async def test_wait_for_timeout_less_then_0_or_0_coroutine_do_not_started(self):
foo_started = False
async def foo():
nonlocal foo_started
foo_started = True
with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(foo(), 0)
self.assertEqual(foo_started, False)
async def test_wait_for_timeout_less_then_0_or_0(self):
loop = asyncio.get_running_loop()
for timeout in [0, -1]:
with self.subTest(timeout=timeout):
foo_running = None
started = loop.create_future()
async def foo():
nonlocal foo_running
foo_running = True
started.set_result(None)
try:
await asyncio.sleep(10)
finally:
foo_running = False
return 'done'
fut = asyncio.create_task(foo())
await started
with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(fut, timeout)
self.assertTrue(fut.done())
# it should have been cancelled due to the timeout
self.assertTrue(fut.cancelled())
self.assertEqual(foo_running, False)
async def test_wait_for(self):
foo_running = None
async def foo():
nonlocal foo_running
foo_running = True
try:
await asyncio.sleep(support.LONG_TIMEOUT)
finally:
foo_running = False
return 'done'
fut = asyncio.create_task(foo())
with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(fut, 0.1)
self.assertTrue(fut.done())
# it should have been cancelled due to the timeout
self.assertTrue(fut.cancelled())
self.assertEqual(foo_running, False)
async def test_wait_for_blocking(self):
async def coro():
return 'done'
res = await asyncio.wait_for(coro(), timeout=None)
self.assertEqual(res, 'done')
async def test_wait_for_race_condition(self):
loop = asyncio.get_running_loop()
fut = loop.create_future()
task = asyncio.wait_for(fut, timeout=0.2)
loop.call_soon(fut.set_result, "ok")
res = await task
self.assertEqual(res, "ok")
async def test_wait_for_cancellation_race_condition(self):
async def inner():
with self.assertRaises(asyncio.CancelledError):
await asyncio.sleep(1)
return 1
result = await asyncio.wait_for(inner(), timeout=.01)
self.assertEqual(result, 1)
async def test_wait_for_waits_for_task_cancellation(self):
task_done = False
async def inner():
nonlocal task_done
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
await asyncio.sleep(_EPSILON)
raise
finally:
task_done = True
inner_task = asyncio.create_task(inner())
with self.assertRaises(asyncio.TimeoutError) as cm:
await asyncio.wait_for(inner_task, timeout=_EPSILON)
self.assertTrue(task_done)
chained = cm.exception.__context__
self.assertEqual(type(chained), asyncio.CancelledError)
async def test_wait_for_waits_for_task_cancellation_w_timeout_0(self):
task_done = False
async def foo():
async def inner():
nonlocal task_done
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
await asyncio.sleep(_EPSILON)
raise
finally:
task_done = True
inner_task = asyncio.create_task(inner())
await asyncio.sleep(_EPSILON)
await asyncio.wait_for(inner_task, timeout=0)
with self.assertRaises(asyncio.TimeoutError) as cm:
await foo()
self.assertTrue(task_done)
chained = cm.exception.__context__
self.assertEqual(type(chained), asyncio.CancelledError)
async def test_wait_for_reraises_exception_during_cancellation(self):
class FooException(Exception):
pass
async def foo():
async def inner():
try:
await asyncio.sleep(0.2)
finally:
raise FooException
inner_task = asyncio.create_task(inner())
await asyncio.wait_for(inner_task, timeout=_EPSILON)
with self.assertRaises(FooException):
await foo()
async def _test_cancel_wait_for(self, timeout):
loop = asyncio.get_running_loop()
async def blocking_coroutine():
fut = loop.create_future()
# Block: fut result is never set
await fut
task = asyncio.create_task(blocking_coroutine())
wait = asyncio.create_task(asyncio.wait_for(task, timeout))
loop.call_soon(wait.cancel)
with self.assertRaises(asyncio.CancelledError):
await wait
# Python issue #23219: cancelling the wait must also cancel the task
self.assertTrue(task.cancelled())
async def test_cancel_blocking_wait_for(self):
await self._test_cancel_wait_for(None)
async def test_cancel_wait_for(self):
await self._test_cancel_wait_for(60.0)
async def test_wait_for_cancel_suppressed(self):
# GH-86296: Suppressing CancelledError is discouraged
# but if a task suppresses CancelledError and returns a value,
# `wait_for` should return the value instead of raising CancelledError.
# This is the same behavior as `asyncio.timeout`.
async def return_42():
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
return 42
res = await asyncio.wait_for(return_42(), timeout=0.1)
self.assertEqual(res, 42)
async def test_wait_for_issue86296(self):
# GH-86296: The task should get cancelled and not run to completion.
# inner completes in one cycle of the event loop so it
# completes before the task is cancelled.
async def inner():
return 'done'
inner_task = asyncio.create_task(inner())
reached_end = False
async def wait_for_coro():
await asyncio.wait_for(inner_task, timeout=100)
await asyncio.sleep(1)
nonlocal reached_end
reached_end = True
task = asyncio.create_task(wait_for_coro())
self.assertFalse(task.done())
# Run the task
await asyncio.sleep(0)
task.cancel()
with self.assertRaises(asyncio.CancelledError):
await task
self.assertTrue(inner_task.done())
self.assertEqual(await inner_task, 'done')
self.assertFalse(reached_end)
class WaitForShieldTests(unittest.IsolatedAsyncioTestCase):
async def test_zero_timeout(self):
# `asyncio.shield` creates a new task which wraps the passed in
# awaitable and shields it from cancellation so with timeout=0
# the task returned by `asyncio.shield` aka shielded_task gets
# cancelled immediately and the task wrapped by it is scheduled
# to run.
async def coro():
await asyncio.sleep(0.01)
return 'done'
task = asyncio.create_task(coro())
with self.assertRaises(asyncio.TimeoutError):
shielded_task = asyncio.shield(task)
await asyncio.wait_for(shielded_task, timeout=0)
# Task is running in background
self.assertFalse(task.done())
self.assertFalse(task.cancelled())
self.assertTrue(shielded_task.cancelled())
# Wait for the task to complete
await asyncio.sleep(0.1)
self.assertTrue(task.done())
async def test_none_timeout(self):
# With timeout=None the timeout is disabled so it
# runs till completion.
async def coro():
await asyncio.sleep(0.1)
return 'done'
task = asyncio.create_task(coro())
await asyncio.wait_for(asyncio.shield(task), timeout=None)
self.assertTrue(task.done())
self.assertEqual(await task, "done")
async def test_shielded_timeout(self):
# shield prevents the task from being cancelled.
async def coro():
await asyncio.sleep(0.1)
return 'done'
task = asyncio.create_task(coro())
with self.assertRaises(asyncio.TimeoutError):
await asyncio.wait_for(asyncio.shield(task), timeout=0.01)
self.assertFalse(task.done())
self.assertFalse(task.cancelled())
self.assertEqual(await task, "done")
if __name__ == '__main__':
unittest.main()
|
