-
Notifications
You must be signed in to change notification settings - Fork 258
/
test_jsonrpc.py
91 lines (70 loc) · 3.09 KB
/
test_jsonrpc.py
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
import uuid
from tornado.testing import AsyncTestCase
from tornado.escape import json_encode, json_decode
from ws4py.client.tornadoclient import TornadoWebSocketClient
import gns3server.jsonrpc as jsonrpc
"""
Tests for JSON-RPC protocol over Websockets
"""
class JSONRPC(AsyncTestCase):
URL = "ws://127.0.0.1:8000/"
def test_request(self):
params = {"echo": "test"}
request = jsonrpc.JSONRPCRequest("dynamips.echo", params)
AsyncWSRequest(self.URL, self.io_loop, self.stop, str(request))
response = self.wait()
json_response = json_decode(response)
assert json_response["jsonrpc"] == 2.0
assert json_response["id"] == request.id
assert json_response["result"] == params
def test_request_with_invalid_method(self):
message = {"echo": "test"}
request = jsonrpc.JSONRPCRequest("dynamips.non_existent", message)
AsyncWSRequest(self.URL, self.io_loop, self.stop, str(request))
response = self.wait()
json_response = json_decode(response)
assert json_response["error"].get("code") == -32601
assert json_response["id"] == request.id
def test_request_with_invalid_version(self):
request = {"jsonrpc": 1.0, "method": "dynamips.echo", "id": 1}
AsyncWSRequest(self.URL, self.io_loop, self.stop, json_encode(request))
response = self.wait()
json_response = json_decode(response)
assert json_response["id"] is None
assert json_response["error"].get("code") == -32600
def test_request_with_invalid_json(self):
request = "my non JSON request"
AsyncWSRequest(self.URL, self.io_loop, self.stop, request)
response = self.wait()
json_response = json_decode(response)
assert json_response["id"] is None
assert json_response["error"].get("code") == -32700
def test_request_with_invalid_jsonrpc_field(self):
request = {"jsonrpc": "2.0", "method_bogus": "dynamips.echo", "id": 1}
AsyncWSRequest(self.URL, self.io_loop, self.stop, json_encode(request))
response = self.wait()
json_response = json_decode(response)
assert json_response["id"] is None
assert json_response["error"].get("code") == -32700
def test_request_with_no_params(self):
request = jsonrpc.JSONRPCRequest("dynamips.echo")
AsyncWSRequest(self.URL, self.io_loop, self.stop, str(request))
response = self.wait()
json_response = json_decode(response)
assert json_response["id"] == request.id
assert json_response["error"].get("code") == -32602
class AsyncWSRequest(TornadoWebSocketClient):
"""
Very basic Websocket client for tests
"""
def __init__(self, url, io_loop, callback, message):
TornadoWebSocketClient.__init__(self, url, io_loop=io_loop)
self._callback = callback
self._message = message
self.connect()
def opened(self):
self.send(self._message, binary=False)
def received_message(self, message):
self.close()
if self._callback:
self._callback(message.data)