| """ |
| 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() |