path: root/tests.py

"""
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.

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


try:
    import json
except ImportError:
    import simplejson as json
import os
import socket
import sys
import tempfile
from threading import Thread

if sys.version_info < (2, 7):
    import unittest2 as unittest
else:
    import unittest

from jsonrpclib import Server, MultiCall, history, ProtocolError
from jsonrpclib import jsonrpc
from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCServer
from jsonrpclib.SimpleJSONRPCServer import SimpleJSONRPCRequestHandler


ORIGINAL_HISTORY_SIZE = history.size


def get_port(family=socket.AF_INET):
    sock = socket.socket(family, socket.SOCK_STREAM)
    sock.bind(("localhost", 0))
    return sock.getsockname()[1]


class TestCompatibility(unittest.TestCase):

    client = None
    port = None
    server = None

    def setUp(self):
        self.port = get_port()
        self.server = server_set_up(addr=('', self.port))
        self.client = Server('http://localhost:%d' % self.port)

    def tearDown(self):
        self.server.stop()
        self.server.join()

    # 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 is 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):
        with 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 "id" in verify_request:
                    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 "error" in verify_response:
                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 = get_port()
        self.server = server_set_up(addr=('', self.port))
        self.addCleanup(self.cleanup)

    def cleanup(self):
        self.server.stop()
        self.server.join()
        history.size = ORIGINAL_HISTORY_SIZE
        history.clear()

    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()
        with 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 is None)

    def test_single_namespace(self):
        client = self.get_client()

        response = client.namespace.sum(1, 2, 4)
        self.assertEqual(7, response)

        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.assertEqual(verify_request, request)
        self.assertEqual(verify_response, response)

    def test_history_defaults_to_20(self):
        client = self.get_client()
        self.assertEqual(20, history.size)

        for i in range(30):
            client.namespace.sum(i, i)

        self.assertEqual(20, len(history.requests))
        self.assertEqual(20, len(history.responses))

        verify_request = {
            "jsonrpc": "2.0", "params": [29, 29],
            "method": "namespace.sum"
        }
        verify_response = {"jsonrpc": "2.0", "result": 58}

        # it should truncate to the *last* 20
        request = json.loads(history.request)
        response = json.loads(history.response)

        verify_request["id"] = request["id"]
        self.assertEqual(request, verify_request)

        verify_response["id"] = request["id"]
        self.assertEqual(response, verify_response)

    def test_history_allows_configurable_size(self):
        client = self.get_client()
        history.size = 10

        for i in range(30):
            client.namespace.sum(i, i)

        self.assertEqual(10, len(history.requests))
        self.assertEqual(10, len(history.responses))

    def test_history_allows_unlimited_size(self):
        client = self.get_client()
        history.size = -1

        for i in range(40):
            client.namespace.sum(i, i)

        self.assertEqual(40, len(history.requests))
        self.assertEqual(40, len(history.responses))

    def test_history_can_be_disabled(self):
        client = self.get_client()
        history.size = 0

        for i in range(40):
            client.namespace.sum(i, i)

        self.assertEqual(0, len(history.requests))
        self.assertEqual(0, len(history.responses))

    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_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]
                with self.assertRaises(raises[i]):
                    func()

    def test_proxy_object_reuse_is_allowed(self):
        client = self.get_client()
        sub_service_proxy = client.sub_service
        result = sub_service_proxy.subtract(5, 10)
        self.assertTrue(result == -5)
        result = sub_service_proxy.add(21, 21)
        self.assertTrue(result == 42)


@unittest.skipIf(
    not jsonrpc.USE_UNIX_SOCKETS or "SKIP_UNIX_SOCKET_TESTS" in os.environ,
    "Skipping Unix socket tests -- unsupported in this environment.")
class UnixSocketInternalTests(InternalTests):
    """
    These tests run the same internal communication tests,
    but over a Unix socket instead of a TCP socket.
    """
    def setUp(self):
        super().setUp()
        suffix = "%d.sock" % get_port()

        # 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 """
        self.server.stop()
        self.server.join()
        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"
        with self.assertRaises(jsonrpc.UnixSocketMissing):
            Server(address)

    def tearDown(self):
        jsonrpc.USE_UNIX_SOCKETS = self.original_value


class ExampleService(object):
    @staticmethod
    def subtract(minuend, subtrahend):
        """ Using the keywords from the JSON-RPC v2 doc """
        return minuend-subtrahend

    @staticmethod
    def add(x, y):
        return x + y

    @staticmethod
    def update(*args):
        return args

    @staticmethod
    def summation(*args):
        return sum(args)

    @staticmethod
    def notify_hello(*args):
        return args

    @staticmethod
    def get_data():
        return ['hello', 5]

    @staticmethod
    def ping():
        return True


class ExampleAggregateService(ExampleService):
    """
    Exposes the inherited ExampleService and a second copy as sub_service
    """
    def __init__(self):
        self.sub_service = ExampleService()


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)
    service = ExampleAggregateService()
    # Expose an instance of the service
    server.register_instance(service, allow_dotted_names=True)
    # Expose some aliases for service methods
    server.register_function(service.summation, 'sum')
    server.register_function(service.summation, 'notify_sum')
    server.register_function(service.summation, 'namespace.sum')

    def stop():
        server.shutdown()

    server_proc = Thread(target=server.serve_forever)
    server_proc.daemon = True
    server_proc.stop = stop
    server_proc.start()
    return server_proc