-
:mod:`nntplib` --- NNTP protocol client
=======================================
reader-specific commands, such as ``group``. If you get unexpected
:exc:`NNTPPermanentError`\ s, you might need to set *readermode*.
:class:`NNTP` class supports the :keyword:`with` statement to
- unconditionally consume :exc:`socket.error` exceptions and to close the NNTP
+ unconditionally consume :exc:`OSError` exceptions and to close the NNTP
connection when done. Here is a sample on how using it:
>>> from nntplib import NNTP
try:
data = self.recv (self.ac_in_buffer_size)
- except socket.error as why:
+ except OSError as why:
self.handle_error()
return
# send the data
try:
num_sent = self.send(data)
- except socket.error:
+ except OSError:
self.handle_error()
return
obj.handle_expt_event()
if flags & (select.POLLHUP | select.POLLERR | select.POLLNVAL):
obj.handle_close()
- except socket.error as e:
+ except OSError as e:
if e.args[0] not in _DISCONNECTED:
obj.handle_error()
else:
# passed be connected.
try:
self.addr = sock.getpeername()
- except socket.error as err:
+ except OSError as err:
if err.args[0] in (ENOTCONN, EINVAL):
# To handle the case where we got an unconnected
# socket.
self.socket.getsockopt(socket.SOL_SOCKET,
socket.SO_REUSEADDR) | 1
)
- except socket.error:
+ except OSError:
pass
# ==================================================
self.addr = address
self.handle_connect_event()
else:
- raise socket.error(err, errorcode[err])
+ raise OSError(err, errorcode[err])
def accept(self):
# XXX can return either an address pair or None
conn, addr = self.socket.accept()
except TypeError:
return None
- except socket.error as why:
+ except OSError as why:
if why.args[0] in (EWOULDBLOCK, ECONNABORTED, EAGAIN):
return None
else:
try:
result = self.socket.send(data)
return result
- except socket.error as why:
+ except OSError as why:
if why.args[0] == EWOULDBLOCK:
return 0
elif why.args[0] in _DISCONNECTED:
return b''
else:
return data
- except socket.error as why:
+ except OSError as why:
# winsock sometimes raises ENOTCONN
if why.args[0] in _DISCONNECTED:
self.handle_close()
self.del_channel()
try:
self.socket.close()
- except socket.error as why:
+ except OSError as why:
if why.args[0] not in (ENOTCONN, EBADF):
raise
def handle_connect_event(self):
err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
if err != 0:
- raise socket.error(err, _strerror(err))
+ raise OSError(err, _strerror(err))
self.handle_connect()
self.connected = True
self.connecting = False
http.putheader('Authorization', auth)
http.endheaders()
http.send(body)
- except socket.error as e:
+ except OSError as e:
self.announce(str(e), log.ERROR)
return
if self.sock is not None:
try:
self.quit()
- except (socket.error, EOFError):
+ except (OSError, EOFError):
pass
finally:
if self.sock is not None:
try:
sock = socket.socket(af, socktype, proto)
sock.bind(sa)
- except socket.error as _:
+ except OSError as _:
err = _
if sock:
sock.close()
if err is not None:
raise err
else:
- raise socket.error("getaddrinfo returns an empty list")
- raise socket.error(msg)
+ raise OSError("getaddrinfo returns an empty list")
+ raise OSError(msg)
sock.listen(1)
port = sock.getsockname()[1] # Get proper port
host = self.sock.getsockname()[0] # Get proper host
if code != 200:
self.close()
- raise socket.error("Tunnel connection failed: %d %s" % (code,
- message.strip()))
+ raise OSError("Tunnel connection failed: %d %s" % (code,
+ message.strip()))
while True:
line = response.fp.readline(_MAXLINE + 1)
if len(line) > _MAXLINE:
try:
self.rpcclt = MyRPCClient(addr)
break
- except socket.error as err:
+ except OSError as err:
pass
else:
self.display_port_binding_error()
raise
except KeyboardInterrupt:
raise
- except socket.error:
+ except OSError:
raise
except Exception as ex:
return ("CALLEXC", ex)
n = self.sock.send(s[:BUFSIZE])
except (AttributeError, TypeError):
raise IOError("socket no longer exists")
- except socket.error:
+ except OSError:
raise
else:
s = s[n:]
return None
try:
s = self.sock.recv(BUFSIZE)
- except socket.error:
+ except OSError:
raise EOFError
if len(s) == 0:
raise EOFError
SocketIO.__init__(self, working_sock)
else:
print("** Invalid host: ", address, file=sys.__stderr__)
- raise socket.error
+ raise OSError
def get_remote_proxy(self, oid):
return RPCProxy(self, oid)
try:
server = MyRPCServer(address, MyHandler)
break
- except socket.error as err:
- print("IDLE Subprocess: socket error: " + err.args[1] +
+ except OSError as err:
+ print("IDLE Subprocess: OSError: " + err.args[1] +
", retrying....", file=sys.__stderr__)
socket_error = err
else:
except Exception:
try:
self.shutdown()
- except socket.error:
+ except OSError:
pass
raise
self.file.close()
try:
self.sock.shutdown(socket.SHUT_RDWR)
- except socket.error as e:
+ except OSError as e:
# The server might already have closed the connection
if e.errno != errno.ENOTCONN:
raise
try:
self.send(data + CRLF)
- except (socket.error, OSError) as val:
+ except OSError as val:
raise self.abort('socket error: %s' % val)
if literal is None:
try:
self.send(literal)
self.send(CRLF)
- except (socket.error, OSError) as val:
+ except OSError as val:
raise self.abort('socket error: %s' % val)
if not literator:
traceback.print_exc()
if self.server.ready:
self.server.ready.set()
- except socket.error as e:
+ except OSError as e:
if not isinstance(e.args, tuple):
raise
else:
try:
self.sock = self.makeSocket()
self.retryTime = None # next time, no delay before trying
- except socket.error:
+ except OSError:
#Creation failed, so set the retry time and return.
if self.retryTime is None:
self.retryPeriod = self.retryStart
if self.sock:
try:
self.sock.sendall(s)
- except socket.error: #pragma: no cover
+ except OSError: #pragma: no cover
self.sock.close()
self.sock = None # so we can call createSocket next time
self.socket = socket.socket(socket.AF_UNIX, self.socktype)
try:
self.socket.connect(address)
- except socket.error:
+ except OSError:
self.socket.close()
raise
if self.unixsocket:
try:
self.socket.send(msg)
- except socket.error:
+ except OSError:
self._connect_unixsocket(self.address)
self.socket.send(msg)
elif self.socktype == socket.SOCK_DGRAM:
if is_connected():
try:
self.quit()
- except (socket.error, EOFError):
+ except (OSError, EOFError):
pass
finally:
if is_connected():
return default
try:
return socket.gethostname()
- except socket.error:
+ except OSError:
# Still not working...
return default
if self.sock is not None:
try:
self.sock.shutdown(socket.SHUT_RDWR)
- except socket.error as e:
+ except OSError as e:
# The server might already have closed the connection
if e.errno != errno.ENOTCONN:
raise
self.num_bytes = 0
try:
self.peer = conn.getpeername()
- except socket.error as err:
+ except OSError as err:
# a race condition may occur if the other end is closing
# before we can get the peername
self.close()
except smtplib.SMTPRecipientsRefused as e:
print('got SMTPRecipientsRefused', file=DEBUGSTREAM)
refused = e.recipients
- except (socket.error, smtplib.SMTPException) as e:
+ except (OSError, smtplib.SMTPException) as e:
print('got', e.__class__, file=DEBUGSTREAM)
# All recipients were refused. If the exception had an associated
# error code, use it. Otherwise,fake it with a non-triggering
"""
try:
request, client_address = self.get_request()
- except socket.error:
+ except OSError:
return
if self.verify_request(request, client_address):
try:
#explicitly shutdown. socket.close() merely releases
#the socket and waits for GC to perform the actual close.
request.shutdown(socket.SHUT_WR)
- except socket.error:
+ except OSError:
pass #some platforms may raise ENOTCONN here
self.close_request(request)
_PROTOCOL_NAMES[PROTOCOL_SSLv2] = "SSLv2"
from socket import getnameinfo as _getnameinfo
-from socket import error as socket_error
from socket import socket, AF_INET, SOCK_STREAM, create_connection
import base64 # for DER-to-PEM translation
import traceback
import errno
+
+socket_error = OSError # keep that public name in module namespace
+
if _ssl.HAS_TLS_UNIQUE:
CHANNEL_BINDING_TYPES = ['tls-unique']
else:
# see if it's connected
try:
sock.getpeername()
- except socket_error as e:
+ except OSError as e:
if e.errno != errno.ENOTCONN:
raise
else:
raise ValueError("do_handshake_on_connect should not be specified for non-blocking sockets")
self.do_handshake()
- except socket_error as x:
+ except OSError as x:
self.close()
raise x
self.do_handshake()
self._connected = True
return rc
- except socket_error:
+ except OSError:
self._sslobj = None
raise
"""Write a string to the socket, doubling any IAC characters.
Can block if the connection is blocked. May raise
- socket.error if the connection is closed.
+ OSError if the connection is closed.
"""
if IAC in buffer:
try:
sock, addr = self.socket.accept()
sslconn = self.context.wrap_socket(sock, server_side=True)
- except socket.error as e:
+ except OSError as e:
# socket errors are silenced by the caller, print them here
if support.verbose:
sys.stderr.write("Got an error:\n%s\n" % e)
the SO_REUSEADDR socket option having different semantics on Windows versus
Unix/Linux. On Unix, you can't have two AF_INET SOCK_STREAM sockets bind,
listen and then accept connections on identical host/ports. An EADDRINUSE
- socket.error will be raised at some point (depending on the platform and
+ OSError will be raised at some point (depending on the platform and
the order bind and listen were called on each socket).
However, on Windows, if SO_REUSEADDR is set on the sockets, no EADDRINUSE
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
sock.bind(('::1', 0))
return True
- except (socket.error, socket.gaierror):
+ except OSError:
pass
finally:
if sock:
# with the Internet connection manifest themselves as exceptions.
# XXX deprecate these and use transient_internet() instead
time_out = TransientResource(IOError, errno=errno.ETIMEDOUT)
-socket_peer_reset = TransientResource(socket.error, errno=errno.ECONNRESET)
+socket_peer_reset = TransientResource(OSError, errno=errno.ECONNRESET)
ioerror_peer_reset = TransientResource(IOError, errno=errno.ECONNRESET)
s2 = asyncore.dispatcher()
s2.create_socket(self.family)
# EADDRINUSE indicates the socket was correctly bound
- self.assertRaises(socket.error, s2.bind, (self.addr[0], port))
+ self.assertRaises(OSError, s2.bind, (self.addr[0], port))
def test_set_reuse_addr(self):
if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
sock = socket.socket(self.family)
try:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- except socket.error:
+ except OSError:
unittest.skip("SO_REUSEADDR not supported on this platform")
else:
# if SO_REUSEADDR succeeded for sock we expect asyncore
struct.pack('ii', 1, 0))
try:
s.connect(server.address)
- except socket.error:
+ except OSError:
pass
finally:
s.close()
client.setblocking(False)
try:
client.connect(('127.0.0.1', self.serverSocket.getsockname()[1]))
- except socket.error as e:
+ except OSError as e:
self.assertEqual(e.args[0], errno.EINPROGRESS)
else:
raise AssertionError("Connect should have raised EINPROGRESS")
elif err.args[0] == ssl.SSL_ERROR_EOF:
return self.handle_close()
raise
- except socket.error as err:
+ except OSError as err:
if err.args[0] == errno.ECONNABORTED:
return self.handle_close()
else:
if err.args[0] in (ssl.SSL_ERROR_WANT_READ,
ssl.SSL_ERROR_WANT_WRITE):
return
- except socket.error as err:
+ except OSError as err:
# Any "socket error" corresponds to a SSL_ERROR_SYSCALL return
# from OpenSSL's SSL_shutdown(), corresponding to a
# closed socket condition. See also:
return False
try:
self.client.sendcmd('noop')
- except (socket.error, EOFError):
+ except (OSError, EOFError):
return False
return True
def sendall(self, data):
if self.pipe_trigger in data:
- raise socket.error(errno.EPIPE, "gotcha")
+ raise OSError(errno.EPIPE, "gotcha")
self.data += data
def close(self):
b"Content-Length")
conn = client.HTTPConnection("example.com")
conn.sock = sock
- self.assertRaises(socket.error,
+ self.assertRaises(OSError,
lambda: conn.request("PUT", "/url", "body"))
resp = conn.getresponse()
self.assertEqual(401, resp.status)
client.setblocking(False)
try:
client.connect(('127.0.0.1', serverSocket.getsockname()[1]))
- except socket.error as e:
+ except OSError as e:
self.assertEqual(e.args[0], errno.EINPROGRESS)
else:
#raise AssertionError("Connect should have raised EINPROGRESS")
self.assertEqual(h.facility, h.LOG_USER)
self.assertTrue(h.unixsocket)
h.close()
- except socket.error: # syslogd might not be available
+ except OSError: # syslogd might not be available
pass
for method in ('GET', 'POST', 'PUT'):
if method == 'PUT':
self.num_bytes = 0
try:
self.peer = conn.getpeername()
- except socket.error as err:
+ except OSError as err:
# a race condition may occur if the other end is closing
# before we can get the peername
self.close()
sock, addr = self.socket.accept()
if self.sslctx:
sock = self.sslctx.wrap_socket(sock, server_side=True)
- except socket.error as e:
+ except OSError as e:
# socket errors are silenced by the caller, print them here
sys.stderr.write("Got an error:\n%s\n" % e)
raise
if data:
try:
super(DelegatingUDPRequestHandler, self).finish()
- except socket.error:
+ except OSError:
if not self.server._closed:
raise
return False
try:
server.help()
- except (socket.error, EOFError):
+ except (OSError, EOFError):
return False
return True
elif err.args[0] == ssl.SSL_ERROR_EOF:
return self.handle_close()
raise
- except socket.error as err:
+ except OSError as err:
if err.args[0] == errno.ECONNABORTED:
return self.handle_close()
else:
smtp.send, 'test msg')
def testNonnumericPort(self):
- # check that non-numeric port raises socket.error
+ # check that non-numeric port raises OSError
self.assertRaises(OSError, smtplib.SMTP,
"localhost", "bogus")
self.assertRaises(OSError, smtplib.SMTP,
"""Check whether CAN sockets are supported on this host."""
try:
s = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
- except (AttributeError, socket.error, OSError):
+ except (AttributeError, OSError):
return False
else:
s.close()
self.addCleanup(self.s.close)
try:
self.s.bind((self.interface,))
- except socket.error:
+ except OSError:
self.skipTest('network interface `%s` does not exist' %
self.interface)
self.cli = socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW)
try:
self.cli.bind((self.interface,))
- except socket.error:
+ except OSError:
# skipTest should not be called here, and will be called in the
# server instead
pass
for obj in args]
try:
s = socket.socket(*callargs)
- except socket.error as e:
+ except OSError as e:
# XXX: check errno?
err = str(e)
else:
def testSocketError(self):
# Testing socket module exceptions
msg = "Error raising socket exception (%s)."
- with self.assertRaises(socket.error, msg=msg % 'socket.error'):
- raise socket.error
- with self.assertRaises(socket.error, msg=msg % 'socket.herror'):
+ with self.assertRaises(OSError, msg=msg % 'OSError'):
+ raise OSError
+ with self.assertRaises(OSError, msg=msg % 'socket.herror'):
raise socket.herror
- with self.assertRaises(socket.error, msg=msg % 'socket.gaierror'):
+ with self.assertRaises(OSError, msg=msg % 'socket.gaierror'):
raise socket.gaierror
def testSendtoErrors(self):
hostname = socket.gethostname()
try:
ip = socket.gethostbyname(hostname)
- except socket.error:
+ except OSError:
# Probably name lookup wasn't set up right; skip this test
return
self.assertTrue(ip.find('.') >= 0, "Error resolving host to ip.")
try:
hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
- except socket.error:
+ except OSError:
# Probably a similar problem as above; skip this test
return
all_host_names = [hostname, hname] + aliases
oldhn = socket.gethostname()
try:
socket.sethostname('new')
- except socket.error as e:
+ except OSError as e:
if e.errno == errno.EPERM:
self.skipTest("test should be run as root")
else:
'socket.if_nameindex() not available.')
def testInvalidInterfaceNameIndex(self):
# test nonexistent interface index/name
- self.assertRaises(socket.error, socket.if_indextoname, 0)
- self.assertRaises(socket.error, socket.if_nametoindex, '_DEADBEEF')
+ self.assertRaises(OSError, socket.if_indextoname, 0)
+ self.assertRaises(OSError, socket.if_nametoindex, '_DEADBEEF')
# test with invalid values
self.assertRaises(TypeError, socket.if_nametoindex, 0)
self.assertRaises(TypeError, socket.if_indextoname, '_DEADBEEF')
try:
# On some versions, this crashes the interpreter.
socket.getnameinfo(('x', 0, 0, 0), 0)
- except socket.error:
+ except OSError:
pass
def testNtoH(self):
try:
port = socket.getservbyname(service, 'tcp')
break
- except socket.error:
+ except OSError:
pass
else:
- raise socket.error
+ raise OSError
# Try same call with optional protocol omitted
port2 = socket.getservbyname(service)
eq(port, port2)
# Try udp, but don't barf it it doesn't exist
try:
udpport = socket.getservbyname(service, 'udp')
- except socket.error:
+ except OSError:
udpport = None
else:
eq(udpport, port)
g = lambda a: inet_pton(AF_INET, a)
assertInvalid = lambda func,a: self.assertRaises(
- (socket.error, ValueError), func, a
+ (OSError, ValueError), func, a
)
self.assertEqual(b'\x00\x00\x00\x00', f('0.0.0.0'))
return
f = lambda a: inet_pton(AF_INET6, a)
assertInvalid = lambda a: self.assertRaises(
- (socket.error, ValueError), f, a
+ (OSError, ValueError), f, a
)
self.assertEqual(b'\x00' * 16, f('::'))
from socket import inet_ntoa as f, inet_ntop, AF_INET
g = lambda a: inet_ntop(AF_INET, a)
assertInvalid = lambda func,a: self.assertRaises(
- (socket.error, ValueError), func, a
+ (OSError, ValueError), func, a
)
self.assertEqual('1.0.1.0', f(b'\x01\x00\x01\x00'))
return
f = lambda a: inet_ntop(AF_INET6, a)
assertInvalid = lambda a: self.assertRaises(
- (socket.error, ValueError), f, a
+ (OSError, ValueError), f, a
)
self.assertEqual('::', f(b'\x00' * 16))
# At least for eCos. This is required for the S/390 to pass.
try:
my_ip_addr = socket.gethostbyname(socket.gethostname())
- except socket.error:
+ except OSError:
# Probably name lookup wasn't set up right; skip this test
return
self.assertIn(name[0], ("0.0.0.0", my_ip_addr), '%s invalid' % name[0])
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(1)
sock.close()
- self.assertRaises(socket.error, sock.send, b"spam")
+ self.assertRaises(OSError, sock.send, b"spam")
def testNewAttributes(self):
# testing .family, .type and .protocol
def test_getnameinfo(self):
# only IP addresses are allowed
- self.assertRaises(socket.error, socket.getnameinfo, ('mail.python.org',0), 0)
+ self.assertRaises(OSError, socket.getnameinfo, ('mail.python.org',0), 0)
@unittest.skipUnless(support.is_resource_enabled('network'),
'network is not enabled')
def testTooLongInterfaceName(self):
# most systems limit IFNAMSIZ to 16, take 1024 to be sure
with socket.socket(socket.PF_CAN, socket.SOCK_RAW, socket.CAN_RAW) as s:
- self.assertRaisesRegex(socket.error, 'interface name too long',
+ self.assertRaisesRegex(OSError, 'interface name too long',
s.bind, ('x' * 1024,))
@unittest.skipUnless(hasattr(socket, "CAN_RAW_LOOPBACK"),
self.assertEqual(f, fileno)
# cli_conn cannot be used anymore...
self.assertTrue(self.cli_conn._closed)
- self.assertRaises(socket.error, self.cli_conn.recv, 1024)
+ self.assertRaises(OSError, self.cli_conn.recv, 1024)
self.cli_conn.close()
# ...but we can create another socket using the (still open)
# file descriptor
def _testSendmsgExcessCmsgReject(self):
if not hasattr(socket, "CMSG_SPACE"):
# Can only send one item
- with self.assertRaises(socket.error) as cm:
+ with self.assertRaises(OSError) as cm:
self.sendmsgToServer([MSG], [(0, 0, b""), (0, 0, b"")])
self.assertIsNone(cm.exception.errno)
self.sendToServer(b"done")
def _testSendmsgAfterClose(self):
self.cli_sock.close()
- self.assertRaises(socket.error, self.sendmsgToServer, [MSG])
+ self.assertRaises(OSError, self.sendmsgToServer, [MSG])
class SendmsgStreamTests(SendmsgTests):
@testSendmsgDontWait.client_skip
def _testSendmsgDontWait(self):
try:
- with self.assertRaises(socket.error) as cm:
+ with self.assertRaises(OSError) as cm:
while True:
self.sendmsgToServer([b"a"*512], [], socket.MSG_DONTWAIT)
self.assertIn(cm.exception.errno,
pass
def _testSendmsgNoDestAddr(self):
- self.assertRaises(socket.error, self.cli_sock.sendmsg,
+ self.assertRaises(OSError, self.cli_sock.sendmsg,
[MSG])
- self.assertRaises(socket.error, self.cli_sock.sendmsg,
+ self.assertRaises(OSError, self.cli_sock.sendmsg,
[MSG], [], 0, None)
def testRecvmsgAfterClose(self):
# Check that recvmsg[_into]() fails on a closed socket.
self.serv_sock.close()
- self.assertRaises(socket.error, self.doRecvmsg, self.serv_sock, 1024)
+ self.assertRaises(OSError, self.doRecvmsg, self.serv_sock, 1024)
def _testRecvmsgAfterClose(self):
pass
# call fails, just send msg with no ancillary data.
try:
nbytes = self.sendmsgToServer([msg], ancdata)
- except socket.error as e:
+ except OSError as e:
# Check that it was the system call that failed
self.assertIsInstance(e.errno, int)
nbytes = self.sendmsgToServer([msg])
array.array("i", [self.traffic_class]).tobytes() + b"\x00"),
(socket.IPPROTO_IPV6, socket.IPV6_HOPLIMIT,
array.array("i", [self.hop_limit]))])
- except socket.error as e:
+ except OSError as e:
self.assertIsInstance(e.errno, int)
nbytes = self.sendmsgToServer(
[MSG],
self.serv.settimeout(self.timeout)
def checkInterruptedRecv(self, func, *args, **kwargs):
- # Check that func(*args, **kwargs) raises socket.error with an
+ # Check that func(*args, **kwargs) raises OSError with an
# errno of EINTR when interrupted by a signal.
self.setAlarm(self.alarm_time)
- with self.assertRaises(socket.error) as cm:
+ with self.assertRaises(OSError) as cm:
func(*args, **kwargs)
self.assertNotIsInstance(cm.exception, socket.timeout)
self.assertEqual(cm.exception.errno, errno.EINTR)
def checkInterruptedSend(self, func, *args, **kwargs):
# Check that func(*args, **kwargs), run in a loop, raises
- # socket.error with an errno of EINTR when interrupted by a
+ # OSError with an errno of EINTR when interrupted by a
# signal.
- with self.assertRaises(socket.error) as cm:
+ with self.assertRaises(OSError) as cm:
while True:
self.setAlarm(self.alarm_time)
func(*args, **kwargs)
start = time.time()
try:
self.serv.accept()
- except socket.error:
+ except OSError:
pass
end = time.time()
self.assertTrue((end - start) < 1.0, "Error setting non-blocking mode.")
start = time.time()
try:
self.serv.accept()
- except socket.error:
+ except OSError:
pass
end = time.time()
self.assertTrue((end - start) < 1.0, "Error creating with non-blocking mode.")
self.serv.setblocking(0)
try:
conn, addr = self.serv.accept()
- except socket.error:
+ except OSError:
pass
else:
self.fail("Error trying to do non-blocking accept.")
conn.setblocking(0)
try:
msg = conn.recv(len(MSG))
- except socket.error:
+ except OSError:
pass
else:
self.fail("Error trying to do non-blocking recv.")
self.read_file.close()
self.assertRaises(ValueError, self.read_file.fileno)
self.cli_conn.close()
- self.assertRaises(socket.error, self.cli_conn.getsockname)
+ self.assertRaises(OSError, self.cli_conn.getsockname)
def _testRealClose(self):
pass
@staticmethod
def _raise_eintr():
- raise socket.error(errno.EINTR, "interrupted")
+ raise OSError(errno.EINTR, "interrupted")
def _textiowrap_mock_socket(self, mock, buffering=-1):
raw = socket.SocketIO(mock, "r")
self.assertEqual(msg, self.read_msg)
# ...until the file is itself closed
self.read_file.close()
- self.assertRaises(socket.error, self.cli_conn.recv, 1024)
+ self.assertRaises(OSError, self.cli_conn.recv, 1024)
def _testMakefileClose(self):
self.write_file.write(self.write_msg)
port = support.find_unused_port()
cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.addCleanup(cli.close)
- with self.assertRaises(socket.error) as cm:
+ with self.assertRaises(OSError) as cm:
cli.connect((HOST, port))
self.assertEqual(cm.exception.errno, errno.ECONNREFUSED)
# Issue #9792: errors raised by create_connection() should have
# a proper errno attribute.
port = support.find_unused_port()
- with self.assertRaises(socket.error) as cm:
+ with self.assertRaises(OSError) as cm:
socket.create_connection((HOST, port))
# Issue #16257: create_connection() calls getaddrinfo() against
foo = self.serv.accept()
except socket.timeout:
self.fail("caught timeout instead of error (TCP)")
- except socket.error:
+ except OSError:
ok = True
except:
self.fail("caught unexpected exception (TCP)")
foo = self.serv.recv(1024)
except socket.timeout:
self.fail("caught timeout instead of error (UDP)")
- except socket.error:
+ except OSError:
ok = True
except:
self.fail("caught unexpected exception (UDP)")
class TestExceptions(unittest.TestCase):
def testExceptionTree(self):
- self.assertTrue(issubclass(socket.error, Exception))
- self.assertTrue(issubclass(socket.herror, socket.error))
- self.assertTrue(issubclass(socket.gaierror, socket.error))
- self.assertTrue(issubclass(socket.timeout, socket.error))
+ self.assertTrue(issubclass(OSError, Exception))
+ self.assertTrue(issubclass(socket.herror, OSError))
+ self.assertTrue(issubclass(socket.gaierror, OSError))
+ self.assertTrue(issubclass(socket.timeout, OSError))
class TestLinuxAbstractNamespace(unittest.TestCase):
def testNameOverflow(self):
address = "\x00" + "h" * self.UNIX_PATH_MAX
with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as s:
- self.assertRaises(socket.error, s.bind, address)
+ self.assertRaises(OSError, s.bind, address)
def testStrName(self):
# Check that an abstract name can be passed as a string.
self.assertTrue(sock._closed)
# exception inside with block
with socket.socket() as sock:
- self.assertRaises(socket.error, sock.sendall, b'foo')
+ self.assertRaises(OSError, sock.sendall, b'foo')
self.assertTrue(sock._closed)
def testCreateConnectionBase(self):
with socket.create_connection(address) as sock:
sock.close()
self.assertTrue(sock._closed)
- self.assertRaises(socket.error, sock.sendall, b'foo')
+ self.assertRaises(OSError, sock.sendall, b'foo')
@unittest.skipUnless(hasattr(socket, "SOCK_CLOEXEC"),
def test_wrapped_unconnected(self):
# Methods on an unconnected SSLSocket propagate the original
- # socket.error raise by the underlying socket object.
+ # OSError raise by the underlying socket object.
s = socket.socket(socket.AF_INET)
ss = ssl.wrap_socket(s)
- self.assertRaises(socket.error, ss.recv, 1)
- self.assertRaises(socket.error, ss.recv_into, bytearray(b'x'))
- self.assertRaises(socket.error, ss.recvfrom, 1)
- self.assertRaises(socket.error, ss.recvfrom_into, bytearray(b'x'), 1)
- self.assertRaises(socket.error, ss.send, b'x')
- self.assertRaises(socket.error, ss.sendto, b'x', ('0.0.0.0', 0))
+ self.assertRaises(OSError, ss.recv, 1)
+ self.assertRaises(OSError, ss.recv_into, bytearray(b'x'))
+ self.assertRaises(OSError, ss.recvfrom, 1)
+ self.assertRaises(OSError, ss.recvfrom_into, bytearray(b'x'), 1)
+ self.assertRaises(OSError, ss.send, b'x')
+ self.assertRaises(OSError, ss.sendto, b'x', ('0.0.0.0', 0))
def test_timeout(self):
# Issue #8524: when creating an SSL socket, the timeout of the
sys.stdout.write(" server: read %r (%s), sending back %r (%s)...\n"
% (msg, ctype, msg.lower(), ctype))
self.write(msg.lower())
- except socket.error:
+ except OSError:
if self.server.chatty:
handle_error("Test server failure:\n")
self.close()
return self.handle_close()
except ssl.SSLError:
raise
- except socket.error as err:
+ except OSError as err:
if err.args[0] == errno.ECONNABORTED:
return self.handle_close()
else:
except ssl.SSLError as x:
if support.verbose:
sys.stdout.write("\nSSLError is %s\n" % x.args[1])
- except socket.error as x:
+ except OSError as x:
if support.verbose:
- sys.stdout.write("\nsocket.error is %s\n" % x.args[1])
+ sys.stdout.write("\nOSError is %s\n" % x.args[1])
except IOError as x:
if x.errno != errno.ENOENT:
raise
except ssl.SSLError:
if expect_success:
raise
- except socket.error as e:
+ except OSError as e:
if expect_success or e.errno != errno.ECONNRESET:
raise
else:
if hasattr(ssl, 'PROTOCOL_SSLv2'):
try:
try_protocol_combo(ssl.PROTOCOL_SSLv23, ssl.PROTOCOL_SSLv2, True)
- except (ssl.SSLError, socket.error) as x:
+ except OSError as x:
# this fails on some older versions of OpenSSL (0.9.7l, for instance)
if support.verbose:
sys.stdout.write(
chatty=False) as server:
with socket.socket() as sock:
s = context.wrap_socket(sock)
- with self.assertRaises((OSError, ssl.SSLError)):
+ with self.assertRaises(OSError):
s.connect((HOST, server.port))
self.assertIn("no shared cipher", str(server.conn_errors[0]))
self.data = body
if self.raise_on_endheaders:
import socket
- raise socket.error()
+ raise OSError()
def getresponse(self):
return MockHTTPResponse(MockFile(), {}, 200, "OK")
("Foo", "bar"), ("Spam", "eggs")])
self.assertEqual(http.data, data)
- # check socket.error converted to URLError
+ # check OSError converted to URLError
http.raise_on_endheaders = True
self.assertRaises(urllib.error.URLError, h.do_open, http, req)
xmlrpc.client.ServerProxy('https://localhost:9999').bad_function()
except NotImplementedError:
self.assertFalse(has_ssl, "xmlrpc client's error with SSL support")
- except socket.error:
+ except OSError:
self.assertTrue(has_ssl)
class HelperTestCase(unittest.TestCase):
return True
exc_mess = e.headers.get('X-exception')
except AttributeError:
- # Ignore socket.errors here.
+ # Ignore OSErrors here.
exc_mess = str(e)
if exc_mess and 'temporarily unavailable' in exc_mess.lower():
def make_request_and_skip(self):
try:
xmlrpclib.ServerProxy(URL).my_function()
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ except (xmlrpclib.ProtocolError, OSError) as e:
if not is_unavailable_exception(e):
raise
raise unittest.SkipTest(reason)
try:
p = xmlrpclib.ServerProxy(URL)
self.assertEqual(p.pow(6,8), 6**8)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
p = xmlrpclib.ServerProxy(URL)
self.assertEqual(p.add(start_string, end_string),
start_string + end_string)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
p = xmlrpclib.ServerProxy(URL)
meth = p.system.listMethods()
self.assertEqual(set(meth), expected_methods)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
p = xmlrpclib.ServerProxy(URL)
divhelp = p.system.methodHelp('div')
self.assertEqual(divhelp, 'This is the div function')
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
p = xmlrpclib.ServerProxy(URL)
myfunction = p.system.methodHelp('my_function')
self.assertEqual(myfunction, 'This is my function')
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
p = xmlrpclib.ServerProxy(URL)
divsig = p.system.methodSignature('div')
self.assertEqual(divsig, 'signatures not supported')
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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.assertEqual(add_result, 2+3)
self.assertEqual(pow_result, 6**8)
self.assertEqual(div_result, 127//42)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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.assertEqual(result.results[0]['faultString'],
'<class \'Exception\'>:method "this_is_not_exists" '
'is not supported')
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
try:
p = xmlrpclib.ServerProxy(URL)
self.assertEqual(p.pow(6,8), 6**8)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
try:
p = xmlrpclib.ServerProxy(URL)
p.pow(6,8)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
try:
p = xmlrpclib.ServerProxy(URL)
p.pow(6,8)
- except (xmlrpclib.ProtocolError, socket.error) as e:
+ 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
server = xmlrpclib.ServerProxy("http://time.xmlrpc.com/RPC2")
try:
t0 = server.currentTime.getCurrentTime()
- except socket.error as e:
+ except OSError as e:
self.skipTest("network error: %s" % e)
return
server = xmlrpclib.ServerProxy("http://buildbot.python.org/all/xmlrpc/")
try:
builders = server.getAllBuilders()
- except socket.error as e:
+ except OSError as e:
self.skipTest("network error: %s" % e)
return
self.addCleanup(lambda: server('close')())
try:
h.request(req.get_method(), req.selector, req.data, headers)
- except socket.error as err: # timeout error
+ except OSError as err: # timeout error
h.close()
raise URLError(err)
else:
try:
host = socket.gethostbyname(host)
- except socket.error as msg:
+ except OSError as msg:
raise URLError(msg)
path, attrs = splitattr(req.selector)
dirs = path.split('/')
return getattr(self, name)(url, data)
except (HTTPError, URLError):
raise
- except socket.error as msg:
+ except OSError as msg:
raise IOError('socket error', msg).with_traceback(sys.exc_info()[2])
def open_unknown(self, fullurl, data=None):
try:
hostIP = socket.gethostbyname(hostonly)
hostIP = ip2num(hostIP)
- except socket.error:
+ except OSError:
continue
base = ip2num(m.group(1))
addr = socket.gethostbyname(rawHost)
if addr != rawHost:
host.append(addr)
- except socket.error:
+ except OSError:
pass
try:
fqdn = socket.getfqdn(rawHost)
if fqdn != rawHost:
host.append(fqdn)
- except socket.error:
+ except OSError:
pass
# make a check value list from the registry entry: replace the
# '<local>' string by the localhost entry and the corresponding
# need to PING each one until we find one that's live
try:
s.connect(fn)
- except socket.error:
+ except OSError:
# no good; attempt to clean it out, but don't fail:
try:
os.unlink(fn)
for i in (0, 1):
try:
return self.single_request(host, handler, request_body, verbose)
- except socket.error as e:
- if i or e.errno not in (errno.ECONNRESET, errno.ECONNABORTED, errno.EPIPE):
+ except OSError as e:
+ if i or e.errno not in (errno.ECONNRESET, errno.ECONNABORTED,
+ errno.EPIPE):
raise
except http.client.BadStatusLine: #close after we sent request
if i:
Module interface:
-- socket.error: exception raised for socket specific errors
+- socket.error: exception raised for socket specific errors, alias for OSError
- socket.gaierror: exception raised for getaddrinfo/getnameinfo errors,
a subclass of socket.error
- socket.herror: exception raised for gethostby* errors,
projects / python / commitdiff