def server_view(request, root='/'):
    '''
    Server view handling pingback requests.

    Include this view in your urlconf under any path you like and
    provide a link to this URL in your HTML:

        <link rel="pingback" href="..."/>

    Or send it in an HTTP server header:

        X-Pingback: ...

    The optional parameter "root" sets a root path within which server will
    consider incoming target URLs as its own.
    '''
    try:
        args, method = xmlrpcclient.loads(request.raw_post_data)
        if method != 'pingback.ping':
            raise errors.Error('Unknown method "%s"' % method)
        _handle_pingback(request, root, *args)
        result = xmlrpcclient.dumps(('OK',), methodresponse=True)
    except xmlrpcclient.Fault as fault:
        result = xmlrpcclient.dumps(fault)
    except Exception as e:
        result = xmlrpcclient.dumps(errors.Error(str(e)))
    return http.HttpResponse(result)

def test_xml_loads():
    """
    Test the `ipalib.rpc.xml_loads` function.
    """
    f = rpc.xml_loads
    params = (binary_bytes, utf8_bytes, unicode_str, None)
    wrapped = rpc.xml_wrap(params, API_VERSION)

    # Test un-serializing an RPC request:
    data = dumps(wrapped, 'the_method', allow_none=True)
    (p, m) = f(data)
    assert_equal(m, u'the_method')
    assert_equal(p, params)

    # Test un-serializing an RPC response:
    data = dumps((wrapped,), methodresponse=True, allow_none=True)
    (tup, m) = f(data)
    assert m is None
    assert len(tup) == 1
    assert type(tup) is tuple
    assert_equal(tup[0], params)

    # Test un-serializing an RPC response containing a Fault:
    for error in (unicode_str, u'hello'):
        fault = Fault(69, error)
        data = dumps(fault, methodresponse=True, allow_none=True, encoding='UTF-8')
        e = raises(Fault, f, data)
        assert e.faultCode == 69
        assert_equal(e.faultString, error)
        assert type(e.faultString) is unicode

    def _marshaled_dispatch(self, data, dispatch_method=None):
        """override SimpleXMLRPCDispatcher._marshaled_dispatch() fault string"""

        try:
            import xmlrpc.client as client
            from xmlrpc.client import Fault
        except ImportError:
            import xmlrpclib as client
            from xmlrpclib import Fault

        params, method = client.loads(data)

        # generate response
        try:
            if dispatch_method is not None:
                response = dispatch_method(method, params)
            else:
                response = self._dispatch(method, params)
            # wrap response in a singleton tuple
            response = (response,)
            response = client.dumps(response, methodresponse=1)
        except Fault as fault: # breaks 2.5 compatibility
            fault.faultString = print_exc_info()
            response = client.dumps(fault)
        except:
            # report exception back to server
            response = client.dumps(
                client.Fault(1, "\n%s" % print_exc_info())
                )

        return _b(response)

    def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
        """Dispatches an XML-RPC method from marshalled (XML) data.

        XML-RPC methods are dispatched from the marshalled (XML) data
        using the _dispatch method and the result is returned as
        marshalled data. For backwards compatibility, a dispatch
        function can be provided as an argument (see comment in
        SimpleXMLRPCRequestHandler.do_POST) but overriding the
        existing method through subclassing is the preferred means
        of changing method dispatch behavior.
        """

        try:
            params, method = loads(data, use_builtin_types=self.use_builtin_types)

            # generate response
            if dispatch_method is not None:
                response = dispatch_method(method, params)
            else:
                response = self._dispatch(method, params)
            # wrap response in a singleton tuple
            response = (response,)
            response = dumps(response, methodresponse=1, allow_none=self.allow_none, encoding=self.encoding)
        except Fault as fault:
            response = dumps(fault, allow_none=self.allow_none, encoding=self.encoding)
        except:
            # report exception back to server
            exc_type, exc_value, exc_tb = sys.exc_info()
            response = dumps(
                Fault(1, "%s:%s" % (exc_type, exc_value)), encoding=self.encoding, allow_none=self.allow_none
            )

        return response.encode(self.encoding)

    def _marshaled_dispatch(self, data, dispatch_method = None):
        """Dispatches an XML-RPC method from marshalled (XML) data.
        
        XML-RPC methods are dispatched from the marshalled (XML) data
        using the _dispatch method and the result is returned as
        marshalled data. For backwards compatibility, a dispatch
        function can be provided as an argument (see comment in 
        SimpleXMLRPCRequestHandler.do_POST) but overriding the
        existing method through subclassing is the prefered means
        of changing method dispatch behavior.
        """
        
        params, method = xmlrpc_client.loads(data)

        # generate response
        try:
            if dispatch_method is not None:
                response = dispatch_method(method, params)
            else:                
                response = self._dispatch(method, params)
            # wrap response in a singleton tuple
            response = (response,)
            response = xmlrpc_client.dumps(response, methodresponse=1)
        except Fault as fault:
            response = xmlrpc_client.dumps(fault)
        except:
            # report exception back to server
            ftb = self.debug and '\n'+str(traceback.format_tb(sys.exc_info()[2])) or ''
            response = xmlrpc_client.dumps(
                xmlrpc_client.Fault(1, "%s:%s%s" % (sys.exc_info()[0], sys.exc_info()[1],ftb))
                )

        return response

    def _cbRender(self, result, request, responseFailed=None):
        if responseFailed:
            return

        if isinstance(result, Handler):
            result = result.result
        if not isinstance(result, Fault):
            result = (result,)
        try:
            try:
                content = xmlrpclib.dumps(
                    result, methodresponse=True,
                    allow_none=self.allowNone)
            except Exception as e:
                f = Fault(self.FAILURE, "Can't serialize output: %s" % (e,))
                content = xmlrpclib.dumps(f, methodresponse=True,
                                          allow_none=self.allowNone)

            if isinstance(content, unicode):
                content = content.encode('utf8')
            request.setHeader(
                b"content-length", intToBytes(len(content)))
            request.write(content)
        except:
            log.err()
        request.finish()

    def test_bug_1164912(self):
        d = xmlrpclib.DateTime()
        ((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,), methodresponse=True))
        self.assertIsInstance(new_d.value, str)

        # Check that the output of dumps() is still an 8-bit string
        s = xmlrpclib.dumps((new_d,), methodresponse=True)
        self.assertIsInstance(s, str)

def xmlrpc_marshal(value):
    ismethodresponse = not isinstance(value, xmlrpclib.Fault)
    if ismethodresponse:
        if not isinstance(value, tuple):
            value = (value,)
        body = xmlrpclib.dumps(value,  methodresponse=ismethodresponse)
    else:
        body = xmlrpclib.dumps(value)
    return body

 def parse_responses(self, responses):
     try:
         if isinstance(responses[0], xmlrpclib.Fault):
             return xmlrpclib.dumps(responses[0])
     except IndexError:
         pass
     try:
         response_xml = xmlrpclib.dumps(responses, methodresponse=True)
     except TypeError:
         return self.faults.internal_error()
     return response_xml

    def test_dump_encoding(self):
        value = '\u20ac'
        strg = xmlrpclib.dumps((value,), encoding='iso-8859-15')
        strg = "<?xml version='1.0' encoding='iso-8859-15'?>" + strg
        self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
        strg = strg.encode('iso-8859-15')
        self.assertEqual(xmlrpclib.loads(strg)[0][0], value)

        strg = xmlrpclib.dumps((value,), encoding='iso-8859-15',
                               methodresponse=True)
        self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
        strg = strg.encode('iso-8859-15')
        self.assertEqual(xmlrpclib.loads(strg)[0][0], value)

 def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
     try:
         (params, method) = loads(data, use_builtin_types=self.use_builtin_types)
         if dispatch_method is not None:
             response = dispatch_method(method, params)
         else:
             response = self._dispatch(method, params)
         response = (response,)
         response = dumps(response, methodresponse=1, allow_none=self.allow_none, encoding=self.encoding)
     except Fault as fault:
         response = dumps(fault, allow_none=self.allow_none, encoding=self.encoding)
     except:
         (exc_type, exc_value, exc_tb) = sys.exc_info()
         response = dumps(Fault(1, '%s:%s' % (exc_type, exc_value)), encoding=self.encoding, allow_none=self.allow_none)
     return response.encode(self.encoding)

def respond(body, encoding='utf-8', allow_none=0):
    from xmlrpc.client import Fault, dumps
    if not isinstance(body, Fault):
        body = (body,)
    _set_response(dumps(body, methodresponse=1,
                        encoding=encoding,
                        allow_none=allow_none))

    def send_rpc(self, cmd, in_fd, out_fd, *args, **kwargs):
        xml = _xmlrpc_client.dumps(sum(kwargs.items(), args), cmd)
        self._debug_fn(
            "calling ikiwiki procedure `{0}': [{1}]".format(cmd, repr(xml)))
        # ensure that encoded is a str (bytestring in Python 2, Unicode in 3)
        if str is bytes and not isinstance(xml, str):
            encoded = xml.encode('utf8')
        else:
            encoded = xml
        _IkiWikiExtPluginXMLRPCHandler._write(out_fd, encoded)

        self._debug_fn('reading response from ikiwiki...')

        response = _IkiWikiExtPluginXMLRPCHandler._read(in_fd)
        if str is bytes and not isinstance(response, str):
            xml = response.encode('utf8')
        else:
            xml = response
        self._debug_fn(
            'read response to procedure {0} from ikiwiki: [{1}]'.format(
                cmd, repr(xml)))
        if xml is None:
            # ikiwiki is going down
            self._debug_fn('ikiwiki is going down, and so are we...')
            raise GoingDown()

        data = _xmlrpc_client.loads(xml)[0][0]
        self._debug_fn(
            'parsed data from response to procedure {0}: [{1}]'.format(
                cmd, repr(data)))
        return data

    def __init__(
        self, path, host, method, user=None, password=None, allowNone=False, args=(), canceller=None, useDateTime=False
    ):
        """
        @param method: The name of the method to call.
        @type method: C{str}

        @param allowNone: allow the use of None values in parameters. It's
            passed to the underlying xmlrpclib implementation. Defaults to
            C{False}.
        @type allowNone: C{bool} or C{NoneType}

        @param args: the arguments to pass to the method.
        @type args: C{tuple}

        @param canceller: A 1-argument callable passed to the deferred as the
            canceller callback.
        @type canceller: callable or C{NoneType}
        """
        self.path, self.host = path, host
        self.user, self.password = user, password
        self.payload = payloadTemplate % (method, xmlrpclib.dumps(args, allow_none=allowNone))
        if isinstance(self.payload, unicode):
            self.payload = self.payload.encode("utf8")
        self.deferred = defer.Deferred(canceller)
        self.useDateTime = useDateTime

def xmlrpc_handle_exception_int(e):
    if isinstance(e, odoo.exceptions.UserError):
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_WARNING, odoo.tools.ustr(e.value))
    elif isinstance(e, odoo.exceptions.RedirectWarning):
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_WARNING, str(e))
    elif isinstance(e, odoo.exceptions.MissingError):
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_WARNING, str(e))
    elif isinstance (e, odoo.exceptions.AccessError):
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_ACCESS_ERROR, str(e))
    elif isinstance(e, odoo.exceptions.AccessDenied):
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_ACCESS_DENIED, str(e))
    elif isinstance(e, odoo.exceptions.DeferredException):
        info = e.traceback
        # Which one is the best ?
        formatted_info = "".join(traceback.format_exception(*info))
        #formatted_info = odoo.tools.exception_to_unicode(e) + '\n' + info
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_APPLICATION_ERROR, formatted_info)
    else:
        info = sys.exc_info()
        # Which one is the best ?
        formatted_info = "".join(traceback.format_exception(*info))
        #formatted_info = odoo.tools.exception_to_unicode(e) + '\n' + info
        fault = xmlrpclib.Fault(RPC_FAULT_CODE_APPLICATION_ERROR, formatted_info)

    return xmlrpclib.dumps(fault, allow_none=None)

def wsgi_xmlrpc(environ, start_response):
    """ Two routes are available for XML-RPC

    /xmlrpc/<service> route returns faultCode as strings. This is a historic
    violation of the protocol kept for compatibility.

    /xmlrpc/2/<service> is a new route that returns faultCode as int and is
    therefore fully compliant.
    """
    if environ['REQUEST_METHOD'] == 'POST' and environ['PATH_INFO'].startswith('/xmlrpc/'):
        length = int(environ['CONTENT_LENGTH'])
        data = environ['wsgi.input'].read(length)

        # Distinguish betweed the 2 faultCode modes
        string_faultcode = True
        service = environ['PATH_INFO'][len('/xmlrpc/'):]
        if environ['PATH_INFO'].startswith('/xmlrpc/2/'):
            service = service[len('2/'):]
            string_faultcode = False

        params, method = xmlrpclib.loads(data)
        try:
            result = odoo.http.dispatch_rpc(service, method, params)
            response = xmlrpclib.dumps((result,), methodresponse=1, allow_none=False)
        except Exception as e:
            if string_faultcode:
                response = xmlrpc_handle_exception_string(e)
            else:
                response = xmlrpc_handle_exception_int(e)

        return werkzeug.wrappers.Response(
            response=response,
            mimetype='text/xml',
        )(environ, start_response)

    def test_dump_big_int(self):
        if sys.maxsize > 2 ** 31 - 1:
            self.assertRaises(OverflowError, xmlrpclib.dumps, (int(2 ** 34),))

        xmlrpclib.dumps((xmlrpclib.MAXINT, xmlrpclib.MININT))
        self.assertRaises(OverflowError, xmlrpclib.dumps, (xmlrpclib.MAXINT + 1,))
        self.assertRaises(OverflowError, xmlrpclib.dumps, (xmlrpclib.MININT - 1,))

        def dummy_write(s):
            pass

        m = xmlrpclib.Marshaller()
        m.dump_int(xmlrpclib.MAXINT, dummy_write)
        m.dump_int(xmlrpclib.MININT, dummy_write)
        self.assertRaises(OverflowError, m.dump_int, xmlrpclib.MAXINT + 1, dummy_write)
        self.assertRaises(OverflowError, m.dump_int, xmlrpclib.MININT - 1, dummy_write)

 def test_dump_none(self):
     value = alist + [None]
     arg1 = (alist + [None],)
     strg = xmlrpclib.dumps(arg1, allow_none=True)
     self.assertEqual(value,
                       xmlrpclib.loads(strg)[0][0])
     self.assertRaises(TypeError, xmlrpclib.dumps, (arg1,))

    def test_dump_bare_datetime(self):
        # This checks that an unwrapped datetime.date object can be handled
        # by the marshalling code.  This can't be done via test_dump_load()
        # since with use_builtin_types set to 1 the unmarshaller would create
        # datetime objects for the 'datetime[123]' keys as well
        dt = datetime.datetime(2005, 2, 10, 11, 41, 23)
        self.assertEqual(dt, xmlrpclib.DateTime('20050210T11:41:23'))
        s = xmlrpclib.dumps((dt,))

        result, m = xmlrpclib.loads(s, use_builtin_types=True)
        (newdt,) = result
        self.assertEqual(newdt, dt)
        self.assertIs(type(newdt), datetime.datetime)
        self.assertIsNone(m)

        result, m = xmlrpclib.loads(s, use_builtin_types=False)
        (newdt,) = result
        self.assertEqual(newdt, dt)
        self.assertIs(type(newdt), xmlrpclib.DateTime)
        self.assertIsNone(m)

        result, m = xmlrpclib.loads(s, use_datetime=True)
        (newdt,) = result
        self.assertEqual(newdt, dt)
        self.assertIs(type(newdt), datetime.datetime)
        self.assertIsNone(m)

        result, m = xmlrpclib.loads(s, use_datetime=False)
        (newdt,) = result
        self.assertEqual(newdt, dt)
        self.assertIs(type(newdt), xmlrpclib.DateTime)
        self.assertIsNone(m)

 def test_newstyle_class(self):
     class T(object):
         pass
     t = T()
     t.x = 100
     t.y = "Hello"
     ((t2,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((t,)))
     self.assertEqual(t2, t.__dict__)