Setting fcntl to None if not importable, adding tests module, updating README to markdown.

1 files changed, 394 insertions, 0 deletions

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