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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
|
import pytest
pytest_plugins = "pytester"
def test_aiohttp_plugin(testdir: pytest.Testdir) -> None:
testdir.makepyfile(
"""\
import pytest
from unittest import mock
from aiohttp import web
value = web.AppKey('value', str)
async def hello(request):
return web.Response(body=b'Hello, world')
async def create_app():
app = web.Application()
app.router.add_route('GET', '/', hello)
return app
async def test_hello(aiohttp_client) -> None:
client = await aiohttp_client(await create_app())
resp = await client.get('/')
assert resp.status == 200
text = await resp.text()
assert 'Hello, world' in text
async def test_hello_from_app(aiohttp_client) -> None:
app = web.Application()
app.router.add_get('/', hello)
client = await aiohttp_client(app)
resp = await client.get('/')
assert resp.status == 200
text = await resp.text()
assert 'Hello, world' in text
async def test_hello_with_loop(aiohttp_client) -> None:
client = await aiohttp_client(await create_app())
resp = await client.get('/')
assert resp.status == 200
text = await resp.text()
assert 'Hello, world' in text
async def test_noop() -> None:
pass
async def previous(request):
if request.method == 'POST':
with pytest.deprecated_call(): # FIXME: this isn't actually called
request.app[value] = (await request.post())['value']
return web.Response(body=b'thanks for the data')
else:
v = request.app.get(value, 'unknown')
return web.Response(body='value: {}'.format(v).encode())
def create_stateful_app():
app = web.Application()
app.router.add_route('*', '/', previous)
return app
@pytest.fixture
async def cli(aiohttp_client):
return await aiohttp_client(create_stateful_app())
def test_noncoro() -> None:
assert True
async def test_failed_to_create_client(aiohttp_client) -> None:
def make_app():
raise RuntimeError()
with pytest.raises(RuntimeError):
await aiohttp_client(make_app())
async def test_custom_port_aiohttp_client(aiohttp_client, unused_tcp_port):
client = await aiohttp_client(await create_app(),
server_kwargs={'port': unused_tcp_port})
assert client.port == unused_tcp_port
resp = await client.get('/')
assert resp.status == 200
text = await resp.text()
assert 'Hello, world' in text
async def test_custom_port_test_server(aiohttp_server, unused_tcp_port):
app = await create_app()
server = await aiohttp_server(app, port=unused_tcp_port)
assert server.port == unused_tcp_port
"""
)
result = testdir.runpytest("--asyncio-mode=auto")
result.assert_outcomes(passed=8)
def test_aiohttp_raw_server(testdir: pytest.Testdir) -> None:
testdir.makepyfile(
"""\
import pytest
from aiohttp import web
async def handler(request):
return web.Response(text="OK")
@pytest.fixture
async def server(aiohttp_raw_server):
return await aiohttp_raw_server(handler)
@pytest.fixture
async def cli(aiohttp_client, server):
client = await aiohttp_client(server)
return client
async def test_hello(cli) -> None:
resp = await cli.get('/')
assert resp.status == 200
text = await resp.text()
assert 'OK' in text
"""
)
result = testdir.runpytest("--asyncio-mode=auto")
result.assert_outcomes(passed=1)
def test_aiohttp_client_cls_fixture_custom_client_used(testdir: pytest.Testdir) -> None:
testdir.makepyfile(
"""
import pytest
from aiohttp.web import Application
from aiohttp.test_utils import TestClient
class CustomClient(TestClient):
pass
@pytest.fixture
def aiohttp_client_cls():
return CustomClient
async def test_hello(aiohttp_client) -> None:
client = await aiohttp_client(Application())
assert isinstance(client, CustomClient)
"""
)
result = testdir.runpytest("--asyncio-mode=auto")
result.assert_outcomes(passed=1)
def test_aiohttp_client_cls_fixture_factory(testdir: pytest.Testdir) -> None:
testdir.makeconftest(
"""\
def pytest_configure(config):
config.addinivalue_line("markers", "rest: RESTful API tests")
config.addinivalue_line("markers", "graphql: GraphQL API tests")
"""
)
testdir.makepyfile(
"""
import pytest
from aiohttp.web import Application
from aiohttp.test_utils import TestClient
class RESTfulClient(TestClient):
pass
class GraphQLClient(TestClient):
pass
@pytest.fixture
def aiohttp_client_cls(request):
if request.node.get_closest_marker('rest') is not None:
return RESTfulClient
elif request.node.get_closest_marker('graphql') is not None:
return GraphQLClient
return TestClient
@pytest.mark.rest
async def test_rest(aiohttp_client) -> None:
client = await aiohttp_client(Application())
assert isinstance(client, RESTfulClient)
@pytest.mark.graphql
async def test_graphql(aiohttp_client) -> None:
client = await aiohttp_client(Application())
assert isinstance(client, GraphQLClient)
"""
)
result = testdir.runpytest("--asyncio-mode=auto")
result.assert_outcomes(passed=2)
|
