Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with
or
.
Download ZIP
Tree: 86551ed635
Fetching contributors…

Cannot retrieve contributors at this time

457 lines (396 sloc) 16.651 kB
"""
The tests in this file compare the request and response objects
to the JSON-RPC 2.0 specification document, as well as testing
several internal components of the jsonrpclib library. Run this
module without any parameters to run the tests.
Currently, this is not easily tested with a framework like
nosetests because we spin up a daemon thread running the
the Server, and nosetests (at least in my tests) does not
ever "kill" the thread.
If you are testing jsonrpclib and the module doesn't return to
the command prompt after running the tests, you can hit
"Ctrl-C" (or "Ctrl-Break" on Windows) and that should kill it.
TODO:
* Finish implementing JSON-RPC 2.0 Spec tests
* Implement JSON-RPC 1.0 tests
* Implement JSONClass, History, Config tests
"""
from jsonrpclib import Server, MultiCall, history, config, ProtocolError
from jsonrpclib import jsonrpc
from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCRequestHandler
import socket
import tempfile
import unittest
import os
import time
try:
import json
except ImportError:
import simplejson as json
from threading import Thread
PORTS = range(8000, 8999)
class TestCompatibility(unittest.TestCase):
client = None
port = None
server = None
def setUp(self):
self.port = PORTS.pop()
self.server = server_set_up(addr=('', self.port))
self.client = Server('http://localhost:%d' % self.port)
# v1 tests forthcoming
# Version 2.0 Tests
def test_positional(self):
""" Positional arguments in a single call """
result = self.client.subtract(23, 42)
self.assertTrue(result == -19)
result = self.client.subtract(42, 23)
self.assertTrue(result == 19)
request = json.loads(history.request)
response = json.loads(history.response)
verify_request = {
"jsonrpc": "2.0", "method": "subtract",
"params": [42, 23], "id": request['id']
}
verify_response = {
"jsonrpc": "2.0", "result": 19, "id": request['id']
}
self.assertTrue(request == verify_request)
self.assertTrue(response == verify_response)
def test_named(self):
""" Named arguments in a single call """
result = self.client.subtract(subtrahend=23, minuend=42)
self.assertTrue(result == 19)
result = self.client.subtract(minuend=42, subtrahend=23)
self.assertTrue(result == 19)
request = json.loads(history.request)
response = json.loads(history.response)
verify_request = {
"jsonrpc": "2.0", "method": "subtract",
"params": {"subtrahend": 23, "minuend": 42},
"id": request['id']
}
verify_response = {
"jsonrpc": "2.0", "result": 19, "id": request['id']
}
self.assertTrue(request == verify_request)
self.assertTrue(response == verify_response)
def test_notification(self):
""" Testing a notification (response should be null) """
result = self.client._notify.update(1, 2, 3, 4, 5)
self.assertTrue(result == None)
request = json.loads(history.request)
response = history.response
verify_request = {
"jsonrpc": "2.0", "method": "update", "params": [1,2,3,4,5]
}
verify_response = ''
self.assertTrue(request == verify_request)
self.assertTrue(response == verify_response)
def test_non_existent_method(self):
self.assertRaises(ProtocolError, self.client.foobar)
request = json.loads(history.request)
response = json.loads(history.response)
verify_request = {
"jsonrpc": "2.0", "method": "foobar", "id": request['id']
}
verify_response = {
"jsonrpc": "2.0",
"error":
{"code": -32601, "message": response['error']['message']},
"id": request['id']
}
self.assertTrue(request == verify_request)
self.assertTrue(response == verify_response)
def test_invalid_json(self):
invalid_json = '{"jsonrpc": "2.0", "method": "foobar, '+ \
'"params": "bar", "baz]'
response = self.client._run_request(invalid_json)
response = json.loads(history.response)
verify_response = json.loads(
'{"jsonrpc": "2.0", "error": {"code": -32700,'+
' "message": "Parse error."}, "id": null}'
)
verify_response['error']['message'] = response['error']['message']
self.assertTrue(response == verify_response)
def test_invalid_request(self):
invalid_request = '{"jsonrpc": "2.0", "method": 1, "params": "bar"}'
response = self.client._run_request(invalid_request)
response = json.loads(history.response)
verify_response = json.loads(
'{"jsonrpc": "2.0", "error": {"code": -32600, '+
'"message": "Invalid Request."}, "id": null}'
)
verify_response['error']['message'] = response['error']['message']
self.assertTrue(response == verify_response)
def test_batch_invalid_json(self):
invalid_request = '[ {"jsonrpc": "2.0", "method": "sum", '+ \
'"params": [1,2,4], "id": "1"},{"jsonrpc": "2.0", "method" ]'
response = self.client._run_request(invalid_request)
response = json.loads(history.response)
verify_response = json.loads(
'{"jsonrpc": "2.0", "error": {"code": -32700,'+
'"message": "Parse error."}, "id": null}'
)
verify_response['error']['message'] = response['error']['message']
self.assertTrue(response == verify_response)
def test_empty_array(self):
invalid_request = '[]'
response = self.client._run_request(invalid_request)
response = json.loads(history.response)
verify_response = json.loads(
'{"jsonrpc": "2.0", "error": {"code": -32600, '+
'"message": "Invalid Request."}, "id": null}'
)
verify_response['error']['message'] = response['error']['message']
self.assertTrue(response == verify_response)
def test_nonempty_array(self):
invalid_request = '[1,2]'
request_obj = json.loads(invalid_request)
response = self.client._run_request(invalid_request)
response = json.loads(history.response)
self.assertTrue(len(response) == len(request_obj))
for resp in response:
verify_resp = json.loads(
'{"jsonrpc": "2.0", "error": {"code": -32600, '+
'"message": "Invalid Request."}, "id": null}'
)
verify_resp['error']['message'] = resp['error']['message']
self.assertTrue(resp == verify_resp)
def test_batch(self):
multicall = MultiCall(self.client)
multicall.sum(1,2,4)
multicall._notify.notify_hello(7)
multicall.subtract(42,23)
multicall.foo.get(name='myself')
multicall.get_data()
job_requests = [j.request() for j in multicall._job_list]
job_requests.insert(3, '{"foo": "boo"}')
json_requests = '[%s]' % ','.join(job_requests)
requests = json.loads(json_requests)
responses = self.client._run_request(json_requests)
verify_requests = json.loads("""[
{"jsonrpc": "2.0", "method": "sum", "params": [1,2,4], "id": "1"},
{"jsonrpc": "2.0", "method": "notify_hello", "params": [7]},
{"jsonrpc": "2.0", "method": "subtract", "params": [42,23], "id": "2"},
{"foo": "boo"},
{"jsonrpc": "2.0", "method": "foo.get", "params": {"name": "myself"}, "id": "5"},
{"jsonrpc": "2.0", "method": "get_data", "id": "9"}
]""")
# Thankfully, these are in order so testing is pretty simple.
verify_responses = json.loads("""[
{"jsonrpc": "2.0", "result": 7, "id": "1"},
{"jsonrpc": "2.0", "result": 19, "id": "2"},
{"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request."}, "id": null},
{"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found."}, "id": "5"},
{"jsonrpc": "2.0", "result": ["hello", 5], "id": "9"}
]""")
self.assertTrue(len(requests) == len(verify_requests))
self.assertTrue(len(responses) == len(verify_responses))
responses_by_id = {}
response_i = 0
for i in range(len(requests)):
verify_request = verify_requests[i]
request = requests[i]
response = None
if request.get('method') != 'notify_hello':
req_id = request.get('id')
if verify_request.has_key('id'):
verify_request['id'] = req_id
verify_response = verify_responses[response_i]
verify_response['id'] = req_id
responses_by_id[req_id] = verify_response
response_i += 1
response = verify_response
self.assertTrue(request == verify_request)
for response in responses:
verify_response = responses_by_id.get(response.get('id'))
if verify_response.has_key('error'):
verify_response['error']['message'] = \
response['error']['message']
self.assertTrue(response == verify_response)
def test_batch_notifications(self):
multicall = MultiCall(self.client)
multicall._notify.notify_sum(1, 2, 4)
multicall._notify.notify_hello(7)
result = multicall()
self.assertTrue(len(result) == 0)
valid_request = json.loads(
'[{"jsonrpc": "2.0", "method": "notify_sum", '+
'"params": [1,2,4]},{"jsonrpc": "2.0", '+
'"method": "notify_hello", "params": [7]}]'
)
request = json.loads(history.request)
self.assertTrue(len(request) == len(valid_request))
for i in range(len(request)):
req = request[i]
valid_req = valid_request[i]
self.assertTrue(req == valid_req)
self.assertTrue(history.response == '')
class InternalTests(unittest.TestCase):
"""
These tests verify that the client and server portions of
jsonrpclib talk to each other properly.
"""
client = None
server = None
port = None
def setUp(self):
self.port = PORTS.pop()
self.server = server_set_up(addr=('', self.port))
def get_client(self):
return Server('http://localhost:%d' % self.port)
def get_multicall_client(self):
server = self.get_client()
return MultiCall(server)
def test_connect(self):
client = self.get_client()
result = client.ping()
self.assertTrue(result)
def test_single_args(self):
client = self.get_client()
result = client.add(5, 10)
self.assertTrue(result == 15)
def test_single_kwargs(self):
client = self.get_client()
result = client.add(x=5, y=10)
self.assertTrue(result == 15)
def test_single_kwargs_and_args(self):
client = self.get_client()
self.assertRaises(ProtocolError, client.add, (5,), {'y':10})
def test_single_notify(self):
client = self.get_client()
result = client._notify.add(5, 10)
self.assertTrue(result == None)
def test_single_namespace(self):
client = self.get_client()
response = client.namespace.sum(1,2,4)
request = json.loads(history.request)
response = json.loads(history.response)
verify_request = {
"jsonrpc": "2.0", "params": [1, 2, 4],
"id": "5", "method": "namespace.sum"
}
verify_response = {
"jsonrpc": "2.0", "result": 7, "id": "5"
}
verify_request['id'] = request['id']
verify_response['id'] = request['id']
self.assertTrue(verify_request == request)
self.assertTrue(verify_response == response)
def test_multicall_success(self):
multicall = self.get_multicall_client()
multicall.ping()
multicall.add(5, 10)
multicall.namespace.sum([5, 10, 15])
correct = [True, 15, 30]
i = 0
for result in multicall():
self.assertTrue(result == correct[i])
i += 1
def test_multicall_success(self):
multicall = self.get_multicall_client()
for i in range(3):
multicall.add(5, i)
result = multicall()
self.assertTrue(result[2] == 7)
def test_multicall_failure(self):
multicall = self.get_multicall_client()
multicall.ping()
multicall.add(x=5, y=10, z=10)
raises = [None, ProtocolError]
result = multicall()
for i in range(2):
if not raises[i]:
result[i]
else:
def func():
return result[i]
self.assertRaises(raises[i], func)
if jsonrpc.USE_UNIX_SOCKETS:
# We won't do these tests unless Unix Sockets are supported
class UnixSocketInternalTests(InternalTests):
"""
These tests run the same internal communication tests,
but over a Unix socket instead of a TCP socket.
"""
def setUp(self):
suffix = "%d.sock" % PORTS.pop()
# Open to safer, alternative processes
# for getting a temp file name...
temp = tempfile.NamedTemporaryFile(
suffix=suffix
)
self.port = temp.name
temp.close()
self.server = server_set_up(
addr=self.port,
address_family=socket.AF_UNIX
)
def get_client(self):
return Server('unix:/%s' % self.port)
def tearDown(self):
""" Removes the tempory socket file """
os.unlink(self.port)
class UnixSocketErrorTests(unittest.TestCase):
"""
Simply tests that the proper exceptions fire if
Unix sockets are attempted to be used on a platform
that doesn't support them.
"""
def setUp(self):
self.original_value = jsonrpc.USE_UNIX_SOCKETS
if (jsonrpc.USE_UNIX_SOCKETS):
jsonrpc.USE_UNIX_SOCKETS = False
def test_client(self):
address = "unix://shouldnt/work.sock"
self.assertRaises(
jsonrpc.UnixSocketMissing,
Server,
address
)
def tearDown(self):
jsonrpc.USE_UNIX_SOCKETS = self.original_value
""" Test Methods """
def subtract(minuend, subtrahend):
""" Using the keywords from the JSON-RPC v2 doc """
return minuend-subtrahend
def add(x, y):
return x + y
def update(*args):
return args
def summation(*args):
return sum(args)
def notify_hello(*args):
return args
def get_data():
return ['hello', 5]
def ping():
return True
def server_set_up(addr, address_family=socket.AF_INET):
# Not sure this is a good idea to spin up a new server thread
# for each test... but it seems to work fine.
def log_request(self, *args, **kwargs):
""" Making the server output 'quiet' """
pass
SimpleJSONRPCRequestHandler.log_request = log_request
server = SimpleJSONRPCServer(addr, address_family=address_family)
server.register_function(summation, 'sum')
server.register_function(summation, 'notify_sum')
server.register_function(notify_hello)
server.register_function(subtract)
server.register_function(update)
server.register_function(get_data)
server.register_function(add)
server.register_function(ping)
server.register_function(summation, 'namespace.sum')
server_proc = Thread(target=server.serve_forever)
server_proc.daemon = True
server_proc.start()
return server_proc
if __name__ == '__main__':
print "==============================================================="
print " NOTE: There may be threading exceptions after tests finish. "
print "==============================================================="
time.sleep(2)
unittest.main()
Jump to Line
Something went wrong with that request. Please try again.