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
|
import asyncio
import sys
import pytest
from async_interrupt import _Interrupt, interrupt
@pytest.mark.asyncio
async def test_interrupt_immediate():
loop = asyncio.get_running_loop()
future = loop.create_future()
with pytest.raises(ValueError, match="message"):
async with interrupt(future, ValueError, "message"):
future.set_result(None)
await asyncio.sleep(0)
@pytest.mark.asyncio
async def test_interrupt_immediate_no_message():
loop = asyncio.get_running_loop()
future = loop.create_future()
with pytest.raises(ValueError):
async with interrupt(future, ValueError, None):
future.set_result(None)
await asyncio.sleep(0)
@pytest.mark.asyncio
async def test_interrupt_soon():
loop = asyncio.get_running_loop()
future = loop.create_future()
with pytest.raises(ValueError, match="message"):
async with interrupt(future, ValueError, "message"):
loop.call_soon(future.set_result, None)
await asyncio.sleep(1)
@pytest.mark.asyncio
async def test_interrupt_later():
loop = asyncio.get_running_loop()
future = loop.create_future()
with pytest.raises(ValueError, match="message"):
async with interrupt(future, ValueError, "message"):
loop.call_later(0.001, future.set_result, None)
await asyncio.sleep(1)
@pytest.mark.asyncio
async def test_interrupt_does_not_happen():
loop = asyncio.get_running_loop()
future = loop.create_future()
async with interrupt(future, ValueError, "message"):
await asyncio.sleep(0.001)
@pytest.mark.asyncio
async def test_interrupt_does_not_happen_after_context_exit():
loop = asyncio.get_running_loop()
future = loop.create_future()
loop.call_soon(future.set_result, None)
async with interrupt(future, ValueError, "message"):
await asyncio.sleep(0)
@pytest.mark.asyncio
async def test_reuse_not_allowed():
loop = asyncio.get_running_loop()
future = loop.create_future()
loop.call_soon(future.set_result, None)
manual_interrupt = _Interrupt(future, ValueError, "message")
async with manual_interrupt:
await asyncio.sleep(0)
with pytest.raises(RuntimeError):
async with manual_interrupt:
await asyncio.sleep(0)
@pytest.mark.skipif(
sys.version_info < (3, 11),
reason="py3.11 is the first version to support uncancel()",
)
@pytest.mark.asyncio
async def test_interrupt_with_current_task_canceled():
loop = asyncio.get_running_loop()
future = loop.create_future()
with pytest.raises(asyncio.CancelledError):
async with interrupt(future, ValueError, "message"):
if task := asyncio.current_task():
task.cancel("external cancel")
future.set_result(None)
await asyncio.sleep(1)
|
