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
|
import asyncio
import pytest
from aiormq.tools import Countdown, awaitable
def simple_func():
return 1
def await_result_func():
return asyncio.sleep(0)
async def await_func():
await asyncio.sleep(0)
return 2
def return_future():
loop = asyncio.get_event_loop()
f = loop.create_future()
loop.call_soon(f.set_result, 5)
return f
async def await_future():
return (await return_future()) + 1
def return_coroutine():
return await_future()
AWAITABLE_FUNCS = [
(simple_func, 1),
(await_result_func, None),
(await_func, 2),
(return_future, 5),
(await_future, 6),
(return_coroutine, 6),
]
@pytest.mark.parametrize("func,result", AWAITABLE_FUNCS)
async def test_awaitable(func, result, loop):
assert await awaitable(func)() == result
async def test_countdown(loop):
countdown = Countdown(timeout=0.1)
await countdown(asyncio.sleep(0))
# waiting for the countdown exceeded
await asyncio.sleep(0.2)
task = asyncio.create_task(asyncio.sleep(0))
with pytest.raises(asyncio.TimeoutError):
await countdown(task)
assert task.cancelled()
|
