tests.py

   1 """
   2 The tests in this file compare the request and response objects
   3 to the JSON-RPC 2.0 specification document, as well as testing
   4 several internal components of the jsonrpclib library. Run this
   5 module without any parameters to run the tests.
   6
   7 Currently, this is not easily tested with a framework like
   8 nosetests because we spin up a daemon thread running the
   9 the Server, and nosetests (at least in my tests) does not
  10 ever "kill" the thread.
  11
  12 If you are testing jsonrpclib and the module doesn't return to
  13 the command prompt after running the tests, you can hit
  14 "Ctrl-C" (or "Ctrl-Break" on Windows) and that should kill it.
  15
  16 TODO:
  17 * Finish implementing JSON-RPC 2.0 Spec tests
  18 * Implement JSON-RPC 1.0 tests
  19 * Implement JSONClass, History, Config tests
  20 """
  21
  22 from jsonrpclib import Server, MultiCall, history, config, ProtocolError
  23 from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
  24 from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCRequestHandler
  25 import unittest
  26 import os
  27 try:
  28     import json
  29 except ImportError:
  30     import simplejson as json
  31 from threading import Thread
  32
  33 PORTS = range(8000, 8999)
  34
  35 class TestCompatibility(unittest.TestCase):
  36
  37     client = None
  38     port = None
  39     server = None
  40
  41     def setUp(self):
  42         self.port = PORTS.pop()
  43         self.server = server_set_up(port=self.port)
  44         self.client = Server('http://localhost:%d' % self.port)
  45
  46     # v1 tests forthcoming
  47
  48     # Version 2.0 Tests
  49     def test_positional(self):
  50         """ Positional arguments in a single call """
  51         result = self.client.subtract(23, 42)
  52         self.assertTrue(result == -19)
  53         result = self.client.subtract(42, 23)
  54         self.assertTrue(result == 19)
  55         request = json.loads(history.request)
  56         response = json.loads(history.response)
  57         verify_request = {
  58             "jsonrpc": "2.0", "method": "subtract",
  59             "params": [42, 23], "id": request['id']
  60         }
  61         verify_response = {
  62             "jsonrpc": "2.0", "result": 19, "id": request['id']
  63         }
  64         self.assertTrue(request == verify_request)
  65         self.assertTrue(response == verify_response)
  66
  67     def test_named(self):
  68         """ Named arguments in a single call """
  69         result = self.client.subtract(subtrahend=23, minuend=42)
  70         self.assertTrue(result == 19)
  71         result = self.client.subtract(minuend=42, subtrahend=23)
  72         self.assertTrue(result == 19)
  73         request = json.loads(history.request)
  74         response = json.loads(history.response)
  75         verify_request = {
  76             "jsonrpc": "2.0", "method": "subtract",
  77             "params": {"subtrahend": 23, "minuend": 42},
  78             "id": request['id']
  79         }
  80         verify_response = {
  81             "jsonrpc": "2.0", "result": 19, "id": request['id']
  82         }
  83         self.assertTrue(request == verify_request)
  84         self.assertTrue(response == verify_response)
  85
  86     def test_notification(self):
  87         """ Testing a notification (response should be null) """
  88         result = self.client._notify.update(1, 2, 3, 4, 5)
  89         self.assertTrue(result == None)
  90         request = json.loads(history.request)
  91         response = history.response
  92         verify_request = {
  93             "jsonrpc": "2.0", "method": "update", "params": [1,2,3,4,5]
  94         }
  95         verify_response = ''
  96         self.assertTrue(request == verify_request)
  97         self.assertTrue(response == verify_response)
  98
  99     def test_non_existent_method(self):
 100         self.assertRaises(ProtocolError, self.client.foobar)
 101         request = json.loads(history.request)
 102         response = json.loads(history.response)
 103         verify_request = {
 104             "jsonrpc": "2.0", "method": "foobar", "id": request['id']
 105         }
 106         verify_response = {
 107             "jsonrpc": "2.0",
 108             "error":
 109                 {"code": -32601, "message": response['error']['message']},
 110             "id": request['id']
 111         }
 112         self.assertTrue(request == verify_request)
 113         self.assertTrue(response == verify_response)
 114
 115     def test_invalid_json(self):
 116         invalid_json = '{"jsonrpc": "2.0", "method": "foobar, '+ \
 117             '"params": "bar", "baz]'
 118         response = self.client._run_request(invalid_json)
 119         response = json.loads(history.response)
 120         verify_response = json.loads(
 121             '{"jsonrpc": "2.0", "error": {"code": -32700,'+
 122             ' "message": "Parse error."}, "id": null}'
 123         )
 124         verify_response['error']['message'] = response['error']['message']
 125         self.assertTrue(response == verify_response)
 126
 127     def test_invalid_request(self):
 128         invalid_request = '{"jsonrpc": "2.0", "method": 1, "params": "bar"}'
 129         response = self.client._run_request(invalid_request)
 130         response = json.loads(history.response)
 131         verify_response = json.loads(
 132             '{"jsonrpc": "2.0", "error": {"code": -32600, '+
 133             '"message": "Invalid Request."}, "id": null}'
 134         )
 135         verify_response['error']['message'] = response['error']['message']
 136         self.assertTrue(response == verify_response)
 137
 138     def test_batch_invalid_json(self):
 139         invalid_request = '[ {"jsonrpc": "2.0", "method": "sum", '+ \
 140             '"params": [1,2,4], "id": "1"},{"jsonrpc": "2.0", "method" ]'
 141         response = self.client._run_request(invalid_request)
 142         response = json.loads(history.response)
 143         verify_response = json.loads(
 144             '{"jsonrpc": "2.0", "error": {"code": -32700,'+
 145             '"message": "Parse error."}, "id": null}'
 146         )
 147         verify_response['error']['message'] = response['error']['message']
 148         self.assertTrue(response == verify_response)
 149
 150     def test_empty_array(self):
 151         invalid_request = '[]'
 152         response = self.client._run_request(invalid_request)
 153         response = json.loads(history.response)
 154         verify_response = json.loads(
 155             '{"jsonrpc": "2.0", "error": {"code": -32600, '+
 156             '"message": "Invalid Request."}, "id": null}'
 157         )
 158         verify_response['error']['message'] = response['error']['message']
 159         self.assertTrue(response == verify_response)
 160
 161     def test_nonempty_array(self):
 162         invalid_request = '[1,2]'
 163         request_obj = json.loads(invalid_request)
 164         response = self.client._run_request(invalid_request)
 165         response = json.loads(history.response)
 166         self.assertTrue(len(response) == len(request_obj))
 167         for resp in response:
 168             verify_resp = json.loads(
 169                 '{"jsonrpc": "2.0", "error": {"code": -32600, '+
 170                 '"message": "Invalid Request."}, "id": null}'
 171             )
 172             verify_resp['error']['message'] = resp['error']['message']
 173             self.assertTrue(resp == verify_resp)
 174
 175     def test_batch(self):
 176         multicall = MultiCall(self.client)
 177         multicall.sum(1,2,4)
 178         multicall._notify.notify_hello(7)
 179         multicall.subtract(42,23)
 180         multicall.foo.get(name='myself')
 181         multicall.get_data()
 182         job_requests = [j.request() for j in multicall._job_list]
 183         job_requests.insert(3, '{"foo": "boo"}')
 184         json_requests = '[%s]' % ','.join(job_requests)
 185         requests = json.loads(json_requests)
 186         responses = self.client._run_request(json_requests)
 187
 188         verify_requests = json.loads("""[
 189             {"jsonrpc": "2.0", "method": "sum", "params": [1,2,4], "id": "1"},
 190             {"jsonrpc": "2.0", "method": "notify_hello", "params": [7]},
 191             {"jsonrpc": "2.0", "method": "subtract", "params": [42,23], "id": "2"},
 192             {"foo": "boo"},
 193             {"jsonrpc": "2.0", "method": "foo.get", "params": {"name": "myself"}, "id": "5"},
 194             {"jsonrpc": "2.0", "method": "get_data", "id": "9"}
 195         ]""")
 196
 197         # Thankfully, these are in order so testing is pretty simple.
 198         verify_responses = json.loads("""[
 199             {"jsonrpc": "2.0", "result": 7, "id": "1"},
 200             {"jsonrpc": "2.0", "result": 19, "id": "2"},
 201             {"jsonrpc": "2.0", "error": {"code": -32600, "message": "Invalid Request."}, "id": null},
 202             {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found."}, "id": "5"},
 203             {"jsonrpc": "2.0", "result": ["hello", 5], "id": "9"}
 204         ]""")
 205
 206         self.assertTrue(len(requests) == len(verify_requests))
 207         self.assertTrue(len(responses) == len(verify_responses))
 208
 209         responses_by_id = {}
 210         response_i = 0
 211
 212         for i in range(len(requests)):
 213             verify_request = verify_requests[i]
 214             request = requests[i]
 215             response = None
 216             if request.get('method') != 'notify_hello':
 217                 req_id = request.get('id')
 218                 if verify_request.has_key('id'):
 219                     verify_request['id'] = req_id
 220                 verify_response = verify_responses[response_i]
 221                 verify_response['id'] = req_id
 222                 responses_by_id[req_id] = verify_response
 223                 response_i += 1
 224                 response = verify_response
 225             self.assertTrue(request == verify_request)
 226
 227         for response in responses:
 228             verify_response = responses_by_id.get(response.get('id'))
 229             if verify_response.has_key('error'):
 230                 verify_response['error']['message'] = \
 231                     response['error']['message']
 232             self.assertTrue(response == verify_response)
 233
 234     def test_batch_notifications(self):
 235         multicall = MultiCall(self.client)
 236         multicall._notify.notify_sum(1, 2, 4)
 237         multicall._notify.notify_hello(7)
 238         result = multicall()
 239         self.assertTrue(len(result) == 0)
 240         valid_request = json.loads(
 241             '[{"jsonrpc": "2.0", "method": "notify_sum", '+
 242             '"params": [1,2,4]},{"jsonrpc": "2.0", '+
 243             '"method": "notify_hello", "params": [7]}]'
 244         )
 245         request = json.loads(history.request)
 246         self.assertTrue(len(request) == len(valid_request))
 247         for i in range(len(request)):
 248             req = request[i]
 249             valid_req = valid_request[i]
 250             self.assertTrue(req == valid_req)
 251         self.assertTrue(history.response == '')
 252
 253 class InternalTests(unittest.TestCase):
 254     """
 255     These tests verify that the client and server portions of
 256     jsonrpclib talk to each other properly.
 257     """
 258     client = None
 259     server = None
 260     port = None
 261
 262     def setUp(self):
 263         self.port = PORTS.pop()
 264         self.server = server_set_up(port=self.port)
 265
 266     def get_client(self):
 267         return Server('http://localhost:%d' % self.port)
 268
 269     def get_multicall_client(self):
 270         server = self.get_client()
 271         return MultiCall(server)
 272
 273     def test_connect(self):
 274         client = self.get_client()
 275         result = client.ping()
 276         self.assertTrue(result)
 277
 278     def test_single_args(self):
 279         client = self.get_client()
 280         result = client.add(5, 10)
 281         self.assertTrue(result == 15)
 282
 283     def test_single_kwargs(self):
 284         client = self.get_client()
 285         result = client.add(x=5, y=10)
 286         self.assertTrue(result == 15)
 287
 288     def test_single_kwargs_and_args(self):
 289         client = self.get_client()
 290         self.assertRaises(ProtocolError, client.add, (5,), {'y':10})
 291
 292     def test_single_notify(self):
 293         client = self.get_client()
 294         result = client._notify.add(5, 10)
 295         self.assertTrue(result == None)
 296
 297     def test_single_namespace(self):
 298         client = self.get_client()
 299         response = client.namespace.sum(1,2,4)
 300         request = json.loads(history.request)
 301         response = json.loads(history.response)
 302         verify_request = {
 303             "jsonrpc": "2.0", "params": [1, 2, 4],
 304             "id": "5", "method": "namespace.sum"
 305         }
 306         verify_response = {
 307             "jsonrpc": "2.0", "result": 7, "id": "5"
 308         }
 309         verify_request['id'] = request['id']
 310         verify_response['id'] = request['id']
 311         self.assertTrue(verify_request == request)
 312         self.assertTrue(verify_response == response)
 313
 314     def test_multicall_success(self):
 315         multicall = self.get_multicall_client()
 316         multicall.ping()
 317         multicall.add(5, 10)
 318         multicall.namespace.sum([5, 10, 15])
 319         correct = [True, 15, 30]
 320         i = 0
 321         for result in multicall():
 322             self.assertTrue(result == correct[i])
 323             i += 1
 324
 325     def test_multicall_success(self):
 326         multicall = self.get_multicall_client()
 327         for i in range(3):
 328             multicall.add(5, i)
 329         result = multicall()
 330         self.assertTrue(result[2] == 7)
 331
 332     def test_multicall_failure(self):
 333         multicall = self.get_multicall_client()
 334         multicall.ping()
 335         multicall.add(x=5, y=10, z=10)
 336         raises = [None, ProtocolError]
 337         result = multicall()
 338         for i in range(2):
 339             if not raises[i]:
 340                 result[i]
 341             else:
 342                 def func():
 343                     return result[i]
 344                 self.assertRaises(raises[i], func)
 345
 346
 347 """ Test Methods """
 348 def subtract(minuend, subtrahend):
 349     """ Using the keywords from the JSON-RPC v2 doc """
 350     return minuend-subtrahend
 351
 352 def add(x, y):
 353     return x + y
 354
 355 def update(*args):
 356     return args
 357
 358 def summation(*args):
 359     return sum(args)
 360
 361 def notify_hello(*args):
 362     return args
 363
 364 def get_data():
 365     return ['hello', 5]
 366
 367 def ping():
 368     return True
 369
 370 def server_set_up(port):
 371     # Not sure this is a good idea to spin up a new server thread
 372     # for each test... but it seems to work fine.
 373     def log_request(self, *args, **kwargs):
 374         """ Making the server output 'quiet' """
 375         pass
 376     SimpleJSONRPCRequestHandler.log_request = log_request
 377     server = SimpleJSONRPCServer(('', port))
 378     server.register_function(summation, 'sum')
 379     server.register_function(summation, 'notify_sum')
 380     server.register_function(notify_hello)
 381     server.register_function(subtract)
 382     server.register_function(update)
 383     server.register_function(get_data)
 384     server.register_function(add)
 385     server.register_function(ping)
 386     server.register_function(summation, 'namespace.sum')
 387     server_proc = Thread(target=server.serve_forever)
 388     server_proc.daemon = True
 389     server_proc.start()
 390     return server_proc
 391
 392 if __name__ == '__main__':
 393     unittest.main()
 394