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
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
|
import asyncio
from unittest import mock
import json
import pytest
import aiohttp
import aiohttp.web
import aiohttp.test_utils
import jsonrpc_base
from jsonrpc_async import Server, ProtocolError, TransportError
async def test_send_message_timeout(aiohttp_client):
"""Test the catching of the timeout responses."""
async def handler(request):
try:
await asyncio.sleep(10)
except asyncio.CancelledError:
# Event loop will be terminated before sleep finishes
pass
return aiohttp.web.Response(text='{}', content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler)
return app
client = await aiohttp_client(create_app())
server = Server('/', client, timeout=0.2)
with pytest.raises(TransportError) as transport_error:
await server.send_message(jsonrpc_base.Request(
'my_method', params=None, msg_id=1))
assert isinstance(transport_error.value.args[1], asyncio.TimeoutError)
async def test_send_message(aiohttp_client):
"""Test the sending of messages."""
# catch non-json responses
async def handler1(request):
return aiohttp.web.Response(
text='not json', content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler1)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
with pytest.raises(TransportError) as transport_error:
await server.send_message(
jsonrpc_base.Request('my_method', params=None, msg_id=1))
assert transport_error.value.args[0] == (
"Error calling method 'my_method': Cannot deserialize response body")
assert isinstance(transport_error.value.args[1], ValueError)
# catch non-200 responses
async def handler2(request):
return aiohttp.web.Response(
text='{}', content_type='application/json', status=404)
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler2)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
with pytest.raises(TransportError) as transport_error:
await server.send_message(jsonrpc_base.Request(
'my_method', params=None, msg_id=1))
assert transport_error.value.args[0] == (
"Error calling method 'my_method': HTTP 404 Not Found")
# catch aiohttp own exception
async def callback(*args, **kwargs):
raise aiohttp.ClientOSError('aiohttp exception')
def create_app():
app = aiohttp.web.Application()
return app
client = await aiohttp_client(create_app())
client.post = callback
server = Server('/', client)
with pytest.raises(TransportError) as transport_error:
await server.send_message(jsonrpc_base.Request(
'my_method', params=None, msg_id=1))
assert transport_error.value.args[0] == (
"Error calling method 'my_method': Transport Error")
async def test_exception_passthrough(aiohttp_client):
async def callback(*args, **kwargs):
raise aiohttp.ClientOSError('aiohttp exception')
def create_app():
app = aiohttp.web.Application()
return app
client = await aiohttp_client(create_app())
client.post = callback
server = Server('/', client)
with pytest.raises(TransportError) as transport_error:
await server.foo()
assert transport_error.value.args[0] == (
"Error calling method 'foo': Transport Error")
assert isinstance(transport_error.value.args[1], aiohttp.ClientOSError)
async def test_forbid_private_methods(aiohttp_client):
"""Test that we can't call private methods (those starting with '_')."""
def create_app():
app = aiohttp.web.Application()
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
with pytest.raises(AttributeError):
await server._foo()
# nested private method call
with pytest.raises(AttributeError):
await server.foo.bar._baz()
async def test_headers_passthrough(aiohttp_client):
"""Test that we correctly send RFC headers and merge them with users."""
async def handler(request):
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": true, "id": 1}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler)
return app
client = await aiohttp_client(create_app())
original_post = client.post
async def callback(*args, **kwargs):
expected_headers = {
'Content-Type': 'application/json',
'Accept': 'application/json-rpc',
'X-TestCustomHeader': '1'
}
assert set(expected_headers.items()).issubset(
set(kwargs['headers'].items()))
return await original_post(*args, **kwargs)
client.post = callback
server = Server('/', client, headers={'X-TestCustomHeader': '1'})
await server.foo()
async def test_method_call(aiohttp_client):
"""Mixing *args and **kwargs is forbidden by the spec."""
def create_app():
app = aiohttp.web.Application()
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
with pytest.raises(ProtocolError) as error:
await server.testmethod(1, 2, a=1, b=2)
assert error.value.args[0] == (
"JSON-RPC spec forbids mixing arguments and keyword arguments")
async def test_method_nesting(aiohttp_client):
"""Test that we correctly nest namespaces."""
async def handler(request):
request_message = await request.json()
if (request_message["params"][0] == request_message["method"]):
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": true, "id": 1}',
content_type='application/json')
else:
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": false, "id": 1}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
assert await server.nest.testmethod("nest.testmethod") is True
assert await server.nest.testmethod.some.other.method(
"nest.testmethod.some.other.method") is True
async def test_calls(aiohttp_client):
"""Test RPC call with positional parameters."""
async def handler1(request):
request_message = await request.json()
assert request_message["params"] == [42, 23]
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": 19, "id": 1}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler1)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
assert await server.subtract(42, 23) == 19
async def handler2(request):
request_message = await request.json()
assert request_message["params"] == {'y': 23, 'x': 42}
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": 19, "id": 1}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler2)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
assert await server.subtract(x=42, y=23) == 19
async def handler3(request):
request_message = await request.json()
assert request_message["params"] == [{'foo': 'bar'}]
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": null}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler3)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
await server.foobar({'foo': 'bar'})
async def test_notification(aiohttp_client):
"""Verify that we ignore the server response."""
async def handler(request):
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": 19, "id": 1}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler)
return app
client = await aiohttp_client(create_app())
server = Server('/', client)
assert await server.subtract(42, 23, _notification=True) is None
async def test_custom_loads(aiohttp_client):
"""Test RPC call with custom load."""
loads_mock = mock.Mock(wraps=json.loads)
async def handler(request):
request_message = await request.json()
assert request_message["params"] == [42, 23]
return aiohttp.web.Response(
text='{"jsonrpc": "2.0", "result": 19, "id": 1}',
content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler)
return app
client = await aiohttp_client(create_app())
server = Server('/', client, loads=loads_mock)
assert await server.subtract(42, 23) == 19
assert loads_mock.call_count == 1
async def test_context_manager(aiohttp_client):
# catch non-json responses
async def handler1(request):
return aiohttp.web.Response(
text='not json', content_type='application/json')
def create_app():
app = aiohttp.web.Application()
app.router.add_route('POST', '/', handler1)
return app
client = await aiohttp_client(create_app())
async with Server('/', client) as server:
assert isinstance(server, Server)
assert not server.session.session.closed
assert server.session.session.closed
|
