1
1
"""Quick'n'dirty unit tests for provided fixtures and markers."""
2
2
import asyncio
3
3
import os
4
+ import sys
4
5
import pytest
5
6
6
7
7
- @asyncio .coroutine
8
- def async_coro (loop ):
9
- yield from asyncio .sleep (0 , loop = loop )
8
+ async def async_coro ():
9
+ await asyncio .sleep (0 )
10
10
return 'ok'
11
11
12
12
13
13
def test_event_loop_fixture (event_loop ):
14
14
"""Test the injection of the event_loop fixture."""
15
15
assert event_loop
16
- ret = event_loop .run_until_complete (async_coro (event_loop ))
16
+ ret = event_loop .run_until_complete (async_coro ())
17
17
assert ret == 'ok'
18
18
19
19
20
+ @pytest .mark .skipif (
21
+ sys .version_info >= (3 , 7 ),
22
+ reason = "Default process poll executor is deprecated since Python 3.8"
23
+ )
20
24
def test_event_loop_processpool_fixture (event_loop_process_pool ):
21
25
"""Test the injection of the event_loop with a process pool fixture."""
22
26
assert event_loop_process_pool
23
27
24
28
ret = event_loop_process_pool .run_until_complete (
25
- async_coro (event_loop_process_pool ))
29
+ async_coro ())
26
30
assert ret == 'ok'
27
31
28
32
this_pid = os .getpid ()
@@ -32,50 +36,51 @@ def test_event_loop_processpool_fixture(event_loop_process_pool):
32
36
33
37
34
38
@pytest .mark .asyncio
35
- def test_asyncio_marker ():
39
+ async def test_asyncio_marker ():
36
40
"""Test the asyncio pytest marker."""
37
- yield # sleep(0)
41
+ await asyncio . sleep (0 )
38
42
39
43
40
44
@pytest .mark .asyncio
41
- def test_asyncio_marker_with_default_param (a_param = None ):
45
+ async def test_asyncio_marker_with_default_param (a_param = None ):
42
46
"""Test the asyncio pytest marker."""
43
- yield # sleep(0)
47
+ await asyncio . sleep (0 )
44
48
45
49
50
+ @pytest .mark .skipif (
51
+ sys .version_info >= (3 , 7 ),
52
+ reason = "Default process poll executor is deprecated since Python 3.8"
53
+ )
46
54
@pytest .mark .asyncio_process_pool
47
- def test_asyncio_process_pool_marker (event_loop ):
55
+ async def test_asyncio_process_pool_marker (event_loop ):
48
56
"""Test the asyncio pytest marker."""
49
- ret = yield from async_coro (event_loop )
57
+ ret = await async_coro ()
50
58
assert ret == 'ok'
51
59
52
60
53
61
@pytest .mark .asyncio
54
- def test_unused_port_fixture (unused_tcp_port , event_loop ):
62
+ async def test_unused_port_fixture (unused_tcp_port , event_loop ):
55
63
"""Test the unused TCP port fixture."""
56
64
57
- @asyncio .coroutine
58
- def closer (_ , writer ):
65
+ async def closer (_ , writer ):
59
66
writer .close ()
60
67
61
- server1 = yield from asyncio .start_server (closer , host = 'localhost' ,
62
- port = unused_tcp_port ,
63
- loop = event_loop )
68
+ server1 = await asyncio .start_server (closer , host = 'localhost' ,
69
+ port = unused_tcp_port )
64
70
65
71
with pytest .raises (IOError ):
66
- yield from asyncio .start_server (closer , host = 'localhost' ,
67
- port = unused_tcp_port ,
68
- loop = event_loop )
72
+ await asyncio .start_server (closer , host = 'localhost' ,
73
+ port = unused_tcp_port )
69
74
70
75
server1 .close ()
71
- yield from server1 .wait_closed ()
76
+ await server1 .wait_closed ()
72
77
73
78
74
79
class Test :
75
80
"""Test that asyncio marked functions work in test methods."""
76
81
77
82
@pytest .mark .asyncio
78
- def test_asyncio_marker_method (self , event_loop ):
83
+ async def test_asyncio_marker_method (self , event_loop ):
79
84
"""Test the asyncio pytest marker in a Test class."""
80
- ret = yield from async_coro (event_loop )
85
+ ret = await async_coro ()
81
86
assert ret == 'ok'
0 commit comments