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 render_POST(self, request):
     request.content.seek(0, 0)
     request.setHeader("content-type", "text/xml")
     try:
         if self.useDateTime:
             args, functionPath = xmlrpclib.loads(request.content.read(),
                 use_datetime=True)
         else:
             # Maintain backwards compatibility with Python < 2.5
             args, functionPath = xmlrpclib.loads(request.content.read())
     except Exception as e:
         f = Fault(self.FAILURE, "Can't deserialize input: %s" % (e,))
         self._cbRender(f, request)
     else:
         try:
             function = self.lookupProcedure(functionPath)
         except Fault as f:
             self._cbRender(f, request)
         else:
             # Use this list to track whether the response has failed or not.
             # This will be used later on to decide if the result of the
             # Deferred should be written out and Request.finish called.
             responseFailed = []
             request.notifyFinish().addErrback(responseFailed.append)
             if getattr(function, 'withRequest', False):
                 d = defer.maybeDeferred(function, request, *args)
             else:
                 d = defer.maybeDeferred(function, *args)
             d.addErrback(self._ebRender)
             d.addCallback(self._cbRender, request, responseFailed)
     return server.NOT_DONE_YET

def test_xml_dumps():
    """
    Test the `ipalib.rpc.xml_dumps` function.
    """
    f = rpc.xml_dumps
    params = (binary_bytes, utf8_bytes, unicode_str, None)

    # Test serializing an RPC request:
    data = f(params, API_VERSION, 'the_method')
    (p, m) = loads(data)
    assert_equal(m, u'the_method')
    assert type(p) is tuple
    assert rpc.xml_unwrap(p) == params

    # Test serializing an RPC response:
    data = f((params,), API_VERSION, methodresponse=True)
    (tup, m) = loads(data)
    assert m is None
    assert len(tup) == 1
    assert type(tup) is tuple
    assert rpc.xml_unwrap(tup[0]) == params

    # Test serializing an RPC response containing a Fault:
    fault = Fault(69, unicode_str)
    data = f(fault, API_VERSION, methodresponse=True)
    e = raises(Fault, loads, data)
    assert e.faultCode == 69
    assert_equal(e.faultString, unicode_str)

    def test_datetime_before_1900(self):
        # same as before but with a date before 1900
        dt = datetime.datetime(1,  2, 10, 11, 41, 23)
        s = xmlrpclib.dumps((dt,))
        (newdt,), m = xmlrpclib.loads(s, use_datetime=1)
        self.assertEqual(newdt, dt)
        self.assertEqual(m, None)

        (newdt,), m = xmlrpclib.loads(s, use_datetime=0)
        self.assertEqual(newdt, xmlrpclib.DateTime('00010210T11:41:23'))

    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_datetime 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)
        s = xmlrpclib.dumps((dt,))
        (newdt,), m = xmlrpclib.loads(s, use_datetime=1)
        self.assertEqual(newdt, dt)
        self.assertEqual(m, None)

        (newdt,), m = xmlrpclib.loads(s, use_datetime=0)
        self.assertEqual(newdt, xmlrpclib.DateTime('20050210T11:41:23'))

    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):
        """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 do_POST(self):
        """ Access point from HTTP handler """

        
        def onParent(pid, fromchild, tochild):
            self._server._registerChild(pid, fromchild)
            tochild.write(_b('done\n'))
            tochild.flush()

        def onChild(pid, fromparent, toparent):
            try:
                response = self._server._marshaled_dispatch(data)
                self._sendResponse(response)
                line = _str(fromparent.readline())
                toparent.write(_b('done\n'))
                toparent.flush()
            except:
                logger(name='pathos.xmlrpc', level=30).error(print_exc_info())
            os._exit(0)

        try:
            data = self.rfile.read(int(self.headers['content-length']))
            params, method = client.loads(data)
            if method == 'run': #XXX: _str?
                return spawn2(onParent, onChild)
            else:
                response = self._server._marshaled_dispatch(data)
                self._sendResponse(response)
                return
        except:
            self._debug.error(print_exc_info())
            self.send_response(500)
            self.end_headers()
            return

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 parse_input(data):
    """Parse input data and return a method path and argument tuple

    The data is a string.
    """
    #
    # For example, with the input:
    #
    #   <?xml version="1.0"?>
    #   <methodCall>
    #      <methodName>examples.getStateName</methodName>
    #      <params>
    #         <param>
    #            <value><i4>41</i4></value>
    #            </param>
    #         </params>
    #      </methodCall>
    #
    # the function should return:
    #
    #     ('examples.getStateName', (41,))
    params, method = xmlrpclib.loads(data)
    # Translate '.' to '/' in meth to represent object traversal.
    method = method.replace('.', '/')
    return method, params

    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 process_body():
    """Return (params, method) from request body."""
    try:
        from xmlrpc import client
        return client.loads(cherrypy.request.body.read())
    except Exception:
        return ('ERROR PARAMS', ), 'ERRORMETHOD'

 def parse_request(self, request_body):
     try:
         params, method_name = xmlrpclib.loads(request_body)
     except:
         # Bad request formatting, bad.
         return self.faults.parse_error()
     return ((method_name, params),)

    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 handle(self) -> Union[Response, tuple]:
        """
        Handle a FastRPC request, returns Flask response
        """
        accept_cts = self._get_accepted_content_types()

        if not accept_cts.intersection(self.allowed_content_types):
            logging.warning('No supported content type requested: "%s"', accept_cts)
            return 'Content types in Accept not supported', 400

        if request.headers['Content-Type'] not in self.allowed_content_types:
            logging.warning('Content-Type "%s" is not supported', request.headers['Content-Type'])
            return 'Content-Type not supported', 400

        if fastrpc:
            if FRPC_CONTENT_TYPE == request.headers['Content-Type']:
                # We will be loading binary data, which is sent through chunked transfer encoding - not very friendly.
                # Werkzeug doesn't recognize the header, so the data should be in request.stream BUT
                # since chunked transfer encoding isn't sending Content-Length header, as it does not make sense,
                # werkzeug needs some kind of middleware that handles it. In ideal world, we could use stream
                # because the middleware would set request.environ['wsgi.input_terminated'] - I found none that do that
                # which means for now we'll be supporting just uwsgi until I figure out how to do with with the others
                # like gunicorn etc. big TODO !
                if uwsgi is None:
                    raise NotImplementedError("This application needs to be running on uWSGI, I'm sorry! TODO :) ")
                request_data = uwsgi.chunked_read()
            else:
                request_data = request.data
            args, method_name = fastrpc.loads(request_data)
        else:
            args, method_name = xmlrpc.loads(request.data)

        logging.info('Calling method %s with args: %s', method_name, args)

        return self._create_response(method_name, args, accept_cts)

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_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 _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 test_nil(self):
     body = FauxInstance(public=None)
     faux = FauxResponse()
     response = self._makeOne(faux)
     response.setBody(body)
     data, method = xmlrpclib.loads(faux._body)
     self.assert_(data[0]['public'] is None)

    def test_datetime_before_1900(self):
        # same as before but with a date before 1900
        dt = datetime.datetime(1,  2, 10, 11, 41, 23)
        self.assertEqual(dt, xmlrpclib.DateTime('00010210T11: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)