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