我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用socket.errors()。
def _download_byte_range(self, byte_range, retry_exceptions): # You can occasionally get socket.errors when downloading # chunks from Glacier, so each chunk can be retried up # to 5 times. for _ in range(5): try: response = self.get_output(byte_range) data = response.read() expected_tree_hash = response['TreeHash'] return data, expected_tree_hash except retry_exceptions as e: continue else: raise DownloadArchiveError("There was an error downloading" "byte range %s: %s" % (byte_range, e))
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_basic(self): # check that flag is false by default flagval = xmlrpc.server.SimpleXMLRPCServer._send_traceback_header self.assertEqual(flagval, False) # enable traceback reporting xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True # test a call that shouldn't fail just as a smoke test 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 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 return False
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), 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': '<type \'exceptions.Exception\'>:' # 'method "this_is_not_exists" is not supported'>}] self.assertEqual(result.results[0]['faultCode'], 1) self.assertEqual(result.results[0]['faultString'], '<type \'exceptions.Exception\'>:method "this_is_not_exists" ' 'is not supported') except (xmlrpclib.ProtocolError, socket.error), 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_basic(self): # check that flag is false by default flagval = SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header self.assertEqual(flagval, False) # enable traceback reporting SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True # test a call that shouldn't fail just as a smoke test try: p = xmlrpclib.ServerProxy(URL) self.assertEqual(p.pow(6,8), 6**8) except (xmlrpclib.ProtocolError, socket.error), 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 SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass # Check that errors in the server send back exception/traceback # info when flag is set SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True try: p = xmlrpclib.ServerProxy(URL) p.pow(6,8) except (xmlrpclib.ProtocolError, socket.error), 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 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_introspection2(self): try: # test _methodHelp() p = xmlrpclib.ServerProxy(URL) divhelp = p.system.methodHelp('div') self.assertEqual(divhelp, 'This is the div 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_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 test_simple1(self): try: p = xmlrpclib.ServerProxy(URL) self.assertEqual(p.pow(6,8), 6**8) except (xmlrpclib.ProtocolError, socket.error), 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 = test_support.u(r'P\N{LATIN SMALL LETTER Y WITH CIRCUMFLEX}t') end_string = test_support.u(r'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", "")))
def test_client_encoding(self): start_string = unichr(0x20ac) end_string = unichr(0xa4) try: p = xmlrpclib.ServerProxy(URL, encoding='iso-8859-15') 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", "")))
def test_nonascii_methodname(self): try: p = xmlrpclib.ServerProxy(URL, encoding='iso-8859-15') m = getattr(p, 't\xea\xa8t') self.assertEqual(m(42), 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", ""))) # [ch] The test 404 is causing lots of false alarms.
def test_introspection2(self): try: # test _methodHelp() p = xmlrpclib.ServerProxy(URL) divhelp = p.system.methodHelp('div') self.assertEqual(divhelp, 'This is the div function') except (xmlrpclib.ProtocolError, socket.error), 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), 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), 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 SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass try: p = xmlrpclib.ServerProxy(URL) p.pow(6,8) except (xmlrpclib.ProtocolError, socket.error), 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')
