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
|
import asyncio
import unittest
import time
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')
t0 = loop.time()
ret = await asyncio.wait_for(fut, 0)
t1 = loop.time()
self.assertEqual(ret, 'done')
self.assertTrue(fut.done())
self.assertLess(t1 - t0, 0.1)
async def test_wait_for_timeout_less_then_0_or_0_coroutine_do_not_started(self):
loop = asyncio.get_running_loop()
foo_started = False
async def foo():
nonlocal foo_started
foo_started = True
with self.assertRaises(asyncio.TimeoutError):
t0 = loop.time()
await asyncio.wait_for(foo(), 0)
t1 = loop.time()
self.assertEqual(foo_started, False)
self.assertLess(t1 - t0, 0.1)
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):
t0 = loop.time()
await asyncio.wait_for(fut, timeout)
t1 = loop.time()
self.assertTrue(fut.done())
# it should have been cancelled due to the timeout
self.assertTrue(fut.cancelled())
self.assertEqual(foo_running, False)
self.assertLess(t1 - t0, 0.1)
async def test_wait_for(self):
loop = asyncio.get_running_loop()
foo_running = None
async def foo():
nonlocal foo_running
foo_running = True
try:
await asyncio.sleep(10)
finally:
foo_running = False
return 'done'
fut = asyncio.create_task(foo())
with self.assertRaises(asyncio.TimeoutError):
t0 = loop.time()
await asyncio.wait_for(fut, 0.1)
t1 = loop.time()
self.assertTrue(fut.done())
# it should have been cancelled due to the timeout
self.assertTrue(fut.cancelled())
self.assertLess(t1 - t0, 0.5)
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_later(0.1, 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_wait_for_self_cancellation(self):
async def inner():
try:
await asyncio.sleep(0.3)
except asyncio.CancelledError:
try:
await asyncio.sleep(0.3)
except asyncio.CancelledError:
await asyncio.sleep(0.3)
return 42
inner_task = asyncio.create_task(inner())
wait = asyncio.wait_for(inner_task, timeout=0.1)
# Test that wait_for itself is properly cancellable
# even when the initial task holds up the initial cancellation.
task = asyncio.create_task(wait)
await asyncio.sleep(0.2)
task.cancel()
with self.assertRaises(asyncio.CancelledError):
await task
self.assertEqual(await inner_task, 42)
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)
if __name__ == '__main__':
unittest.main()
|
