我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用xmlrpc.client.ProtocolError()。
def __request(self, methodname, args): retries_left = self.__retries while True: try: return self.__real_request(methodname, args) except (ProtocolError, Fault, socket.error) as exc: if isinstance(exc, Fault): # Retry request on InvalidConfig (a false, infrequent fault) if '[InvalidConfig]' not in exc.faultString: raise if retries_left == 0: raise else: retries_left -= 1 continue
def is_unavailable_exception(e): '''Returns True if the given ProtocolError is the product of a server-side exception caused by the 'temporarily unavailable' response sometimes given by operations on non-blocking sockets.''' # sometimes we get a -1 error code and/or empty headers try: if e.errcode == -1 or e.headers is None: return True exc_mess = e.headers.get('X-exception') except AttributeError: # Ignore socket.errors here. exc_mess = str(e) if exc_mess and 'temporarily unavailable' in exc_mess.lower(): return True
def test_multicall(self): try: p = xmlrpclib.ServerProxy(URL) multicall = xmlrpclib.MultiCall(p) multicall.add(2,3) multicall.pow(6,8) multicall.div(127,42) add_result, pow_result, div_result = multicall() self.assertEqual(add_result, 2+3) self.assertEqual(pow_result, 6**8) self.assertEqual(div_result, 127//42) except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_non_existing_multicall(self): try: p = xmlrpclib.ServerProxy(URL) multicall = xmlrpclib.MultiCall(p) multicall.this_is_not_exists() result = multicall() # result.results contains; # [{'faultCode': 1, 'faultString': '<class \'exceptions.Exception\'>:' # 'method "this_is_not_exists" is not supported'>}] self.assertEqual(result.results[0]['faultCode'], 1) self.assertEqual(result.results[0]['faultString'], '<class \'Exception\'>:method "this_is_not_exists" ' 'is not supported') except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_fail_with_info(self): # use the broken message class xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass # Check that errors in the server send back exception/traceback # info when flag is set xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True try: p = xmlrpclib.ServerProxy(URL) p.pow(6,8) except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e) and hasattr(e, "headers"): # We should get error info in the response expected_err = "invalid literal for int() with base 10: 'I am broken'" self.assertEqual(e.headers.get("X-exception"), expected_err) self.assertTrue(e.headers.get("X-traceback") is not None) else: self.fail('ProtocolError not raised')
def shutdown_server(self): """Send xmlrpc request to shutdown xmlrpc server. """ try: ans = self.xmlproxy.shutdown() except socket_error as err: self.class_logger.info("xmlrpc shutdown complete. (DEBUG: {0})".format(err)) except XmlrpcProtocolError as err: self.class_logger.info("xmlrpc shutdown complete. (DEBUG: {0})".format(err)) except Exception as err: self.class_logger.info("xmlrpc shutdown expected error: {0} - {1}".format(type(err), err)) else: self.class_logger.info("xmlrpc shutdown query answer: %s" % (ans, )) # except socket.error, err: # if err[0] == 111: # print "!"*100 # print "ERR '{0}' handled".format(err) # else: # raise
def is_unavailable_exception(e): '''Returns True if the given ProtocolError is the product of a server-side exception caused by the 'temporarily unavailable' response sometimes given by operations on non-blocking sockets.''' # sometimes we get a -1 error code and/or empty headers try: if e.errcode == -1 or e.headers is None: return True exc_mess = e.headers.get('X-exception') except AttributeError: # Ignore OSErrors here. exc_mess = str(e) if exc_mess and 'temporarily unavailable' in exc_mess.lower(): return True
def test_multicall(self): try: p = xmlrpclib.ServerProxy(URL) multicall = xmlrpclib.MultiCall(p) multicall.add(2,3) multicall.pow(6,8) multicall.div(127,42) add_result, pow_result, div_result = multicall() self.assertEqual(add_result, 2+3) self.assertEqual(pow_result, 6**8) self.assertEqual(div_result, 127//42) except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_non_existing_multicall(self): try: p = xmlrpclib.ServerProxy(URL) multicall = xmlrpclib.MultiCall(p) multicall.this_is_not_exists() result = multicall() # result.results contains; # [{'faultCode': 1, 'faultString': '<class \'exceptions.Exception\'>:' # 'method "this_is_not_exists" is not supported'>}] self.assertEqual(result.results[0]['faultCode'], 1) self.assertEqual(result.results[0]['faultString'], '<class \'Exception\'>:method "this_is_not_exists" ' 'is not supported') except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_fail_with_info(self): # use the broken message class xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass # Check that errors in the server send back exception/traceback # info when flag is set xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True try: p = xmlrpclib.ServerProxy(URL) p.pow(6,8) except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e) and hasattr(e, "headers"): # We should get error info in the response expected_err = "invalid literal for int() with base 10: 'I am broken'" self.assertEqual(e.headers.get("X-exception"), expected_err) self.assertTrue(e.headers.get("X-traceback") is not None) else: self.fail('ProtocolError not raised')
def request(self, host, handler, request_body, verbose=0): # issue XML-RPC request h = self.make_connection(host) if verbose: h.set_debuglevel(1) self.send_request(h, handler, request_body) self.send_host(h, host) self.send_user_agent(h) self.send_content(h, request_body) errcode, errmsg, headers = h.getreply() if errcode != 200: raise ProtocolError(host + handler, errcode, errmsg, headers) self.verbose = verbose # Here's where we differ from the superclass. It says: # try: # sock = h._conn.sock # except AttributeError: # sock = None # return self._parse_response(h.getfile(), sock) return self.parse_response(h.getfile())
def __enter__(self): self._proxy.__enter__() try: methods = self._proxy.system.listMethods() except ProtocolError as e: if e.errcode != 404: raise # remote daemon returned 404, likely using different rmw impl. self._proxy.__exit__() # fall back to use direct node self._proxy = DirectNode(self._args) self._proxy.__enter__() else: self._methods = [m for m in methods if not m.startswith('system.')] return self
def test_pypi_protocol_error(self, _mock): """pypi protocol error.""" _mock.side_effect = xmlrpc_client.ProtocolError(pypi.PYPI_URL, 400, 'dummy', {}) with self.assertRaises(exceptions.BackendFailure): pypi.package_existent('py-deps')
def package_existent(name): """search package. * :class:`bootstrap_py.exceptions.Conflict` exception occurs when user specified name has already existed. * :class:`bootstrap_py.exceptions.BackendFailure` exception occurs when PyPI service is down. :param str name: package name """ if sys.version_info < (3, 0): try: result = search_package(name) except (socket.error, xmlrpc_client.ProtocolError) as exc: raise BackendFailure(exc) else: try: result = search_package(name) except (socket.gaierror, TimeoutError, ConnectionRefusedError, xmlrpc_client.ProtocolError) as exc: raise BackendFailure(exc) if result: msg = ('[error] "{0}" is registered already in PyPI.\n' '\tSpecify another package name.').format(name) raise Conflict(msg)
def make_request_and_skipIf(condition, reason): # If we skip the test, we have to make a request because the # the server created in setUp blocks expecting one to come in. if not condition: return lambda func: func def decorator(func): def make_request_and_skip(self): try: xmlrpclib.ServerProxy(URL).my_function() except (xmlrpclib.ProtocolError, socket.error) as e: if not is_unavailable_exception(e): raise raise unittest.SkipTest(reason) return make_request_and_skip return decorator
def test_simple1(self): try: p = xmlrpclib.ServerProxy(URL) self.assertEqual(p.pow(6,8), 6**8) except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_nonascii(self): start_string = 'P\N{LATIN SMALL LETTER Y WITH CIRCUMFLEX}t' end_string = 'h\N{LATIN SMALL LETTER O WITH HORN}n' try: p = xmlrpclib.ServerProxy(URL) self.assertEqual(p.add(start_string, end_string), start_string + end_string) except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", ""))) # [ch] The test 404 is causing lots of false alarms.
def test_introspection1(self): expected_methods = set(['pow', 'div', 'my_function', 'add', 'system.listMethods', 'system.methodHelp', 'system.methodSignature', 'system.multicall']) try: p = xmlrpclib.ServerProxy(URL) meth = p.system.listMethods() self.assertEqual(set(meth), expected_methods) except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_introspection3(self): try: # test native doc p = xmlrpclib.ServerProxy(URL) myfunction = p.system.methodHelp('my_function') self.assertEqual(myfunction, 'This is my function') except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_introspection4(self): # the SimpleXMLRPCServer doesn't support signatures, but # at least check that we can try making the call try: p = xmlrpclib.ServerProxy(URL) divsig = p.system.methodSignature('div') self.assertEqual(divsig, 'signatures not supported') except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_bad_gzip_request(self): t = self.Transport() t.encode_threshold = None t.fake_gzip = True p = xmlrpclib.ServerProxy(URL, transport=t) cm = self.assertRaisesRegex(xmlrpclib.ProtocolError, re.compile(r"\b400\b")) with cm: p.pow(6, 8) p("close")()
def test_fail_no_info(self): # use the broken message class xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass try: p = xmlrpclib.ServerProxy(URL) p.pow(6,8) except (xmlrpclib.ProtocolError, socket.error) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e) and hasattr(e, "headers"): # The two server-side error headers shouldn't be sent back in this case self.assertTrue(e.headers.get("X-exception") is None) self.assertTrue(e.headers.get("X-traceback") is None) else: self.fail('ProtocolError not raised')
def make_request_and_skipIf(condition, reason): # If we skip the test, we have to make a request because # the server created in setUp blocks expecting one to come in. if not condition: return lambda func: func def decorator(func): def make_request_and_skip(self): try: xmlrpclib.ServerProxy(URL).my_function() except (xmlrpclib.ProtocolError, OSError) as e: if not is_unavailable_exception(e): raise raise unittest.SkipTest(reason) return make_request_and_skip return decorator
def test_simple1(self): try: p = xmlrpclib.ServerProxy(URL) self.assertEqual(p.pow(6,8), 6**8) except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_nonascii(self): start_string = 'P\N{LATIN SMALL LETTER Y WITH CIRCUMFLEX}t' end_string = 'h\N{LATIN SMALL LETTER O WITH HORN}n' try: p = xmlrpclib.ServerProxy(URL) self.assertEqual(p.add(start_string, end_string), start_string + end_string) except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", ""))) # [ch] The test 404 is causing lots of false alarms.
def test_introspection1(self): expected_methods = set(['pow', 'div', 'my_function', 'add', 'system.listMethods', 'system.methodHelp', 'system.methodSignature', 'system.multicall', 'Fixture']) try: p = xmlrpclib.ServerProxy(URL) meth = p.system.listMethods() self.assertEqual(set(meth), expected_methods) except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_introspection3(self): try: # test native doc p = xmlrpclib.ServerProxy(URL) myfunction = p.system.methodHelp('my_function') self.assertEqual(myfunction, 'This is my function') except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_introspection4(self): # the SimpleXMLRPCServer doesn't support signatures, but # at least check that we can try making the call try: p = xmlrpclib.ServerProxy(URL) divsig = p.system.methodSignature('div') self.assertEqual(divsig, 'signatures not supported') except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
def test_fail_no_info(self): # use the broken message class xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass try: p = xmlrpclib.ServerProxy(URL) p.pow(6,8) except (xmlrpclib.ProtocolError, OSError) as e: # ignore failures due to non-blocking socket 'unavailable' errors if not is_unavailable_exception(e) and hasattr(e, "headers"): # The two server-side error headers shouldn't be sent back in this case self.assertTrue(e.headers.get("X-exception") is None) self.assertTrue(e.headers.get("X-traceback") is None) else: self.fail('ProtocolError not raised')
def make_request_and_skipIf(condition, reason): # If we skip the test, we have to make a request because the # the server created in setUp blocks expecting one to come in. if not condition: return lambda func: func def decorator(func): def make_request_and_skip(self): try: xmlrpclib.ServerProxy(URL).my_function() except (xmlrpclib.ProtocolError, OSError) as e: if not is_unavailable_exception(e): raise raise unittest.SkipTest(reason) return make_request_and_skip return decorator
