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
|
# Test asyncio.wait_for, with forwarding cancellation
try:
import asyncio
except ImportError:
print("SKIP")
raise SystemExit
async def awaiting(t, return_if_fail):
try:
print("awaiting started")
await asyncio.sleep(t)
except asyncio.CancelledError as er:
# CPython wait_for raises CancelledError inside task but TimeoutError in wait_for
print("awaiting canceled")
if return_if_fail:
return False # return has no effect if Cancelled
else:
raise er
except Exception as er:
print("caught exception", er)
raise er
async def test_cancellation_forwarded(catch, catch_inside):
print("----------")
async def wait():
try:
await asyncio.wait_for(awaiting(2, catch_inside), 1)
except asyncio.TimeoutError as er:
print("Got timeout error")
raise er
except asyncio.CancelledError as er:
print("Got canceled")
if not catch:
raise er
async def cancel(t):
print("cancel started")
await asyncio.sleep(0.01)
print("cancel wait()")
t.cancel()
t = asyncio.create_task(wait())
k = asyncio.create_task(cancel(t))
try:
await t
except asyncio.CancelledError:
print("waiting got cancelled")
asyncio.run(test_cancellation_forwarded(False, False))
asyncio.run(test_cancellation_forwarded(False, True))
asyncio.run(test_cancellation_forwarded(True, True))
asyncio.run(test_cancellation_forwarded(True, False))
|
