X-Git-Url: http://xvm.mit.edu/gitweb/invirt/packages/python-jsonrpclib.git/blobdiff_plain/283a2a94ca6376d96b2feda716bc3a2ffbb31cb1..6a388ded772221c4b5fbb07ed69db8d57b0b583e:/tests.py diff --git a/tests.py b/tests.py new file mode 100644 index 0000000..c030ffe --- /dev/null +++ b/tests.py @@ -0,0 +1,394 @@ +""" +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.SimpleJSONRPCServer import SimpleJSONRPCServer +from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCRequestHandler +import unittest +import os +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(port=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(port=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) + + +""" 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(port): + # 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(('', port)) + 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__': + unittest.main() +