tests.py

import unittest
import random
import json
import inspect
import os
import sys

import pep8
import requests
import requests.exceptions
import responses

from jsonrpc_requests import Server, ProtocolError, TransportError

try:
    # python >= 3.3
    from unittest.mock import Mock
except ImportError:
    from mock import Mock


class TestCase(unittest.TestCase):
    def assertSameJSON(self, json1, json2):
        """Tells whether two json strings, once decoded, are the same dictionary"""
        return self.assertDictEqual(json.loads(json1), json.loads(json2))

    def assertRaisesRegex(self, *args, **kwargs):
        if hasattr(super(TestCase, self), 'assertRaisesRegex'):
            # python >= 3.3
            return super(TestCase, self).assertRaisesRegex(*args, **kwargs)
        else:
            # python 2.7
            return self.assertRaisesRegexp(*args, **kwargs)


class TestJSONRPCClient(TestCase):

    def setUp(self):
        random.randint = Mock(return_value=1)
        self.server = Server('http://mock/xmlrpc')

    def test_length(self):
        """Verify that this library is really smaller than 100 lines, as stated in README.rst"""
        with open(inspect.getfile(Server)) as library_file:
            self.assertLessEqual(len([l for l in library_file.readlines()
                                      if l.strip()]), 100)

    def test_pep8_conformance(self):
        """Test that we conform to PEP8."""

        source_files = []
        project_dir = os.path.dirname(os.path.abspath(__file__))
        package_dir = os.path.join(project_dir, 'jsonrpc_requests')
        for root, directories, filenames in os.walk(package_dir):
            source_files.extend([os.path.join(root, f) for f in filenames if f.endswith('.py')])

        pep8style = pep8.StyleGuide(quiet=False, max_line_length=120)
        result = pep8style.check_files(source_files)
        self.assertEqual(result.total_errors, 0, "Found code style errors (and warnings).")

    def test_dumps(self):
        # test keyword args
        self.assertSameJSON(
            '''{"params": {"foo": "bar"}, "jsonrpc": "2.0", "method": "my_method_name", "id": 1}''',
            self.server.serialize('my_method_name', params={'foo': 'bar'}, is_notification=False)
        )
        # test positional args
        self.assertSameJSON(
            '''{"params": ["foo", "bar"], "jsonrpc": "2.0", "method": "my_method_name", "id": 1}''',
            self.server.serialize('my_method_name', params=('foo', 'bar'), is_notification=False)
        )
        # test notification
        self.assertSameJSON(
            '''{"params": ["foo", "bar"], "jsonrpc": "2.0", "method": "my_method_name"}''',
            self.server.serialize('my_method_name', params=('foo', 'bar'), is_notification=True)
        )

    @responses.activate
    def test_parse_response(self):
        class MockResponse(object):
            def __init__(self, json_data, status_code=200):
                self.json_data = json_data
                self.status_code = status_code

            def json(self):
                return self.json_data

        # catch non-json responses
        with self.assertRaises(ProtocolError) as protocol_error:
            responses.add(responses.POST, 'http://mock/xmlrpc', body='not json', content_type='application/json')
            self.server.send_request('my_method', is_notification=False, params=None)

        if sys.version_info > (3, 0):
            self.assertEqual(protocol_error.exception.message,
                             """Cannot deserialize response body: Expecting value: line 1 column 1 (char 0)""")
        else:
            self.assertEqual(protocol_error.exception.message,
                             """Cannot deserialize response body: No JSON object could be decoded""")

        self.assertIsInstance(protocol_error.exception.server_response, requests.Response)
        responses.reset()

        with self.assertRaisesRegex(ProtocolError, 'Response is not a dictionary'):
            self.server.parse_response(MockResponse([]))

        with self.assertRaisesRegex(ProtocolError, 'Response without a result field'):
            self.server.parse_response(MockResponse({}))

        with self.assertRaises(ProtocolError) as protoerror:
            body = {"jsonrpc": "2.0", "error": {"code": -32601, "message": "Method not found"}, "id": "1"}
            self.server.parse_response(MockResponse(body))
        self.assertEqual(protoerror.exception.message, '''Error: -32601 Method not found''')

    @responses.activate
    def test_send_request(self):
        # catch non-200 responses
        with self.assertRaisesRegex(TransportError, '404'):
            responses.add(responses.POST, 'http://mock/xmlrpc', body='{}', content_type='application/json', status=404)
            self.server.send_request('my_method', is_notification=False, params=None)
        responses.reset()

        # catch requests own exception
        with self.assertRaisesRegex(TransportError, """Error calling method 'my_method'""") as e:
            # noinspection PyUnusedLocal
            def callback(request):
                raise requests.RequestException('Requests exception')
            responses.add_callback(
                responses.POST, 'http://mock/xmlrpc', content_type='application/json', callback=callback,
            )
            self.server.send_request('my_method', is_notification=False, params=None)
            self.assertIsInstance(e.cause, requests.RequestException)
        responses.reset()

        # a notification
        responses.add(responses.POST, 'http://mock/xmlrpc', body='we dont care about this',
                      content_type='application/json')
        self.server.send_request('my_notification', is_notification=True, params=None)
        responses.reset()

    def test_exception_passthrough(self):
        with self.assertRaises(TransportError) as transport_error:
            s = Server('http://host-doesnt-exist')
            s.foo()
        self.assertEqual(transport_error.exception.message, "Error calling method 'foo'")
        self.assertIsInstance(transport_error.exception.cause, requests.exceptions.RequestException)

    def test_forbid_private_methods(self):
        """Test that we can't call private class methods (those starting with '_')"""
        s = Server('http://host-doesnt-exist')
        with self.assertRaises(AttributeError):
            s._foo()

        # nested private method call
        with self.assertRaises(AttributeError):
            s.foo.bar._baz()

    @responses.activate
    def test_headers_passthrough(self):
        """Test that we correctly send RFC-defined headers and merge them with user defined ones"""
        def callback(request):
            expected_headers = {
                'Content-Type': 'application/json',
                'Accept': 'application/json-rpc',
                'X-TestCustomHeader': '1'
            }
            self.assertTrue(set(expected_headers.items()).issubset(set(request.headers.items())))
            return 200, {}, u'{"jsonrpc": "2.0", "result": true, "id": 1}'

        responses.add_callback(
            responses.POST, 'http://mock/xmlrpc', content_type='application/json', callback=callback,
        )
        s = Server('http://mock/xmlrpc', headers={'X-TestCustomHeader': '1'})
        s.foo()

    def test_method_call(self):
        """mixing *args and **kwargs is forbidden by the spec"""
        with self.assertRaisesRegex(ProtocolError, 'JSON-RPC spec forbids mixing arguments and keyword arguments'):
            self.server.testmethod(1, 2, a=1, b=2)

    @responses.activate
    def test_method_nesting(self):
        """Test that we correctly nest namespaces"""
        def callback(request):
            request_message = json.loads(request.body)
            if request_message["params"][0] == request_message["method"]:
                return 200, {}, u'{"jsonrpc": "2.0", "result": true, "id": 1}'
            else:
                return 200, {}, u'{"jsonrpc": "2.0", "result": false, "id": 1}'

        responses.add_callback(
            responses.POST, 'http://mock/xmlrpc', content_type='application/json', callback=callback,
        )

        self.assertEqual(self.server.nest.testmethod("nest.testmethod"), True)
        self.assertEqual(self.server.nest.testmethod.some.other.method("nest.testmethod.some.other.method"), True)

    @responses.activate
    def test_calls(self):
        # rpc call with positional parameters:
        def callback1(request):
            request_message = json.loads(request.body)
            self.assertEqual(request_message["params"], [42, 23])
            return 200, {}, u'{"jsonrpc": "2.0", "result": 19, "id": 1}'

        responses.add_callback(
            responses.POST, 'http://mock/xmlrpc',
            content_type='application/json',
            callback=callback1,
        )
        self.assertEqual(self.server.subtract(42, 23), 19)
        responses.reset()

        # rpc call with named parameters
        def callback2(request):
            request_message = json.loads(request.body)
            self.assertEqual(request_message["params"], {'y': 23, 'x': 42})
            return 200, {}, u'{"jsonrpc": "2.0", "result": 19, "id": 1}'

        responses.add_callback(
            responses.POST, 'http://mock/xmlrpc',
            content_type='application/json',
            callback=callback2,
        )
        self.assertEqual(self.server.subtract(x=42, y=23), 19)
        responses.reset()

        # rpc call with a mapping type
        def callback3(request):
            request_message = json.loads(request.body)
            self.assertEqual(request_message["params"], {'foo': 'bar'})
            return 200, {}, u'{"jsonrpc": "2.0", "result": null}'

        responses.add_callback(
            responses.POST, 'http://mock/xmlrpc',
            content_type='application/json',
            callback=callback3,
        )
        self.server.foobar({'foo': 'bar'})
        responses.reset()

    @responses.activate
    def test_notification(self):
        # Verify that we ignore the server response
        responses.add(responses.POST, 'http://mock/xmlrpc',
                      body='{"jsonrpc": "2.0", "result": 19, "id": 3}',
                      content_type='application/json')
        self.assertIsNone(self.server.subtract(42, 23, _notification=True))
        responses.reset()


if __name__ == '__main__':
    unittest.main()