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
|
from __future__ import annotations
import asyncio
import math
from asyncio import get_running_loop
from collections.abc import Generator
from unittest import mock
from unittest.mock import AsyncMock
import pytest
from pytest import MonkeyPatch
from anyio import run, sleep_forever, sleep_until
pytestmark = pytest.mark.anyio
fake_current_time = 1620581544.0
@pytest.fixture
def fake_sleep() -> Generator[AsyncMock, None, None]:
with mock.patch(
"anyio._core._eventloop.current_time", return_value=fake_current_time
):
with mock.patch("anyio._core._eventloop.sleep", AsyncMock()) as v:
yield v
async def test_sleep_until(fake_sleep: AsyncMock) -> None:
deadline = fake_current_time + 500.102352
await sleep_until(deadline)
fake_sleep.assert_called_once_with(deadline - fake_current_time)
async def test_sleep_until_in_past(fake_sleep: AsyncMock) -> None:
deadline = fake_current_time - 500.102352
await sleep_until(deadline)
fake_sleep.assert_called_once_with(0)
async def test_sleep_forever(fake_sleep: AsyncMock) -> None:
await sleep_forever()
fake_sleep.assert_called_once_with(math.inf)
def test_run_task() -> None:
"""Test that anyio.run() on asyncio will work with a callable returning a Future."""
async def async_add(x: int, y: int) -> int:
return x + y
result = run(asyncio.create_task, async_add(1, 2), backend="asyncio")
assert result == 3
class TestAsyncioOptions:
def test_debug(self) -> None:
async def main() -> bool:
return get_running_loop().get_debug()
debug = run(main, backend="asyncio", backend_options={"debug": True})
assert debug is True
def test_debug_via_env(self, monkeypatch: MonkeyPatch) -> None:
async def main() -> bool:
return get_running_loop().get_debug()
monkeypatch.setenv("PYTHONASYNCIODEBUG", "1")
debug = run(main, backend="asyncio")
assert debug is True
def test_loop_factory(self) -> None:
async def main() -> type:
return type(get_running_loop())
uvloop = pytest.importorskip("uvloop", reason="uvloop not installed")
loop_class = run(
main,
backend="asyncio",
backend_options={"loop_factory": uvloop.new_event_loop},
)
assert issubclass(loop_class, uvloop.Loop)
def test_use_uvloop(self) -> None:
async def main() -> type:
return type(get_running_loop())
uvloop = pytest.importorskip("uvloop", reason="uvloop not installed")
loop_class = run(main, backend="asyncio", backend_options={"use_uvloop": True})
assert issubclass(loop_class, uvloop.Loop)
|
