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
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
|
import asyncio
import contextlib
import pytest
from aiohttp.web import Application
from .test_utils import unused_port as _unused_port
from .test_utils import (RawTestServer, TestClient, TestServer, loop_context,
setup_test_loop, teardown_test_loop)
@contextlib.contextmanager
def _passthrough_loop_context(loop):
if loop:
# loop already exists, pass it straight through
yield loop
else:
# this shadows loop_context's standard behavior
loop = setup_test_loop()
yield loop
teardown_test_loop(loop)
def pytest_pycollect_makeitem(collector, name, obj):
"""
Fix pytest collecting for coroutines.
"""
if collector.funcnamefilter(name) and asyncio.iscoroutinefunction(obj):
return list(collector._genfunctions(name, obj))
def pytest_pyfunc_call(pyfuncitem):
"""
Run coroutines in an event loop instead of a normal function call.
"""
if asyncio.iscoroutinefunction(pyfuncitem.function):
existing_loop = pyfuncitem.funcargs.get('loop', None)
with _passthrough_loop_context(existing_loop) as _loop:
testargs = {arg: pyfuncitem.funcargs[arg]
for arg in pyfuncitem._fixtureinfo.argnames}
task = _loop.create_task(pyfuncitem.obj(**testargs))
_loop.run_until_complete(task)
return True
@pytest.yield_fixture
def loop():
with loop_context() as _loop:
yield _loop
@pytest.fixture
def unused_port():
return _unused_port
@pytest.yield_fixture
def test_server(loop):
servers = []
@asyncio.coroutine
def go(app, **kwargs):
assert app.loop is loop, \
"Application is attached to other event loop"
server = TestServer(app)
yield from server.start_server(**kwargs)
servers.append(server)
return server
yield go
@asyncio.coroutine
def finalize():
while servers:
yield from servers.pop().close()
loop.run_until_complete(finalize())
@pytest.yield_fixture
def raw_test_server(loop):
servers = []
@asyncio.coroutine
def go(handler, **kwargs):
server = RawTestServer(handler, loop=loop)
yield from server.start_server(**kwargs)
servers.append(server)
return server
yield go
@asyncio.coroutine
def finalize():
while servers:
yield from servers.pop().close()
loop.run_until_complete(finalize())
@pytest.yield_fixture
def test_client(loop):
clients = []
@asyncio.coroutine
def go(__param, *args, **kwargs):
if isinstance(__param, Application):
assert not args, "args should be empty"
assert __param.loop is loop, \
"Application is attached to other event loop"
client = TestClient(__param, **kwargs)
elif isinstance(__param, TestServer):
assert not args, "args should be empty"
assert __param.app.loop is loop, \
"TestServer is attached to other event loop"
client = TestClient(__param, **kwargs)
elif isinstance(__param, RawTestServer):
assert not args, "args should be empty"
assert __param._loop is loop, \
"TestServer is attached to other event loop"
client = TestClient(__param, **kwargs)
else:
__param = __param(loop, *args, **kwargs)
client = TestClient(__param)
yield from client.start_server()
clients.append(client)
return client
yield go
@asyncio.coroutine
def finalize():
while clients:
yield from clients.pop().close()
loop.run_until_complete(finalize())
|
