Repository navigation
Expand file tree
/
Copy pathtest_staggered.py
More file actions
86 lines (67 loc) · 2.28 KB
/
Copy pathtest_staggered.py
File metadata and controls
86 lines (67 loc) · 2.28 KB
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
import asyncio
import sys
from functools import partial
import pytest
from aiohappyeyeballs._staggered import staggered_race
@pytest.mark.asyncio
async def test_one_winners():
"""Test that there is only one winner when there is no await in the coro."""
winners = []
async def coro(idx):
winners.append(idx)
return idx
coros = [partial(coro, idx) for idx in range(4)]
winner, index, excs = await staggered_race(
coros,
delay=None,
)
assert len(winners) == 1
assert winners == [0]
assert winner == 0
assert index == 0
assert excs == [None]
@pytest.mark.asyncio
async def test_multiple_winners():
"""Test multiple winners are handled correctly."""
loop = asyncio.get_running_loop()
winners = []
finish = loop.create_future()
async def coro(idx):
await finish
winners.append(idx)
return idx
coros = [partial(coro, idx) for idx in range(4)]
task = loop.create_task(staggered_race(coros, delay=0.00001))
await asyncio.sleep(0.1)
loop.call_soon(finish.set_result, None)
winner, index, excs = await task
assert len(winners) == 4
assert winners == [0, 1, 2, 3]
assert winner == 0
assert index == 0
assert excs == [None, None, None, None]
@pytest.mark.skipif(sys.version_info < (3, 12), reason="requires python3.12 or higher")
def test_multiple_winners_eager_task_factory():
"""Test multiple winners are handled correctly."""
loop = asyncio.new_event_loop()
eager_task_factory = asyncio.create_eager_task_factory(asyncio.Task)
loop.set_task_factory(eager_task_factory)
asyncio.set_event_loop(None)
async def run():
winners = []
finish = loop.create_future()
async def coro(idx):
await finish
winners.append(idx)
return idx
coros = [partial(coro, idx) for idx in range(4)]
task = loop.create_task(staggered_race(coros, delay=0.00001))
await asyncio.sleep(0.1)
loop.call_soon(finish.set_result, None)
winner, index, excs = await task
assert len(winners) == 4
assert winners == [0, 1, 2, 3]
assert winner == 0
assert index == 0
assert excs == [None, None, None, None]
loop.run_until_complete(run())