Menu
▾
▴
[Jython-checkins] SF.net SVN: jython:[6360] branches/Release_2_2maint/jython/Lib
|
From: <am...@us...> - 2009-05-20 15:09:54
|
Revision: 6360
http://jython.svn.sourceforge.net/jython/?rev=6360&view=rev
Author: amak
Date: 2009-05-20 15:09:43 +0000 (Wed, 20 May 2009)
Log Message:
-----------
Backporting changes from trunk.
1. Socket options are now level specific
2. Explicit backlog parameter on listen
3. Setting has_ipv6 to False
Modified Paths:
--------------
branches/Release_2_2maint/jython/Lib/socket.py
branches/Release_2_2maint/jython/Lib/test/test_socket.py
Modified: branches/Release_2_2maint/jython/Lib/socket.py
===================================================================
--- branches/Release_2_2maint/jython/Lib/socket.py 2009-05-20 06:09:43 UTC (rev 6359)
+++ branches/Release_2_2maint/jython/Lib/socket.py 2009-05-20 15:09:43 UTC (rev 6360)
@@ -141,17 +141,6 @@
SHUT_WR = 1
SHUT_RDWR = 2
-__all__ = ['AF_UNSPEC', 'AF_INET', 'AF_INET6', 'AI_PASSIVE', 'SOCK_DGRAM',
- 'SOCK_RAW', 'SOCK_RDM', 'SOCK_SEQPACKET', 'SOCK_STREAM', 'SOL_SOCKET',
- 'SO_BROADCAST', 'SO_ERROR', 'SO_KEEPALIVE', 'SO_LINGER', 'SO_OOBINLINE',
- 'SO_RCVBUF', 'SO_REUSEADDR', 'SO_SNDBUF', 'SO_TIMEOUT', 'TCP_NODELAY',
- 'SocketType', 'error', 'herror', 'gaierror', 'timeout',
- 'getfqdn', 'gethostbyaddr', 'gethostbyname', 'gethostname',
- 'socket', 'getaddrinfo', 'getdefaulttimeout', 'setdefaulttimeout',
- 'has_ipv6', 'htons', 'htonl', 'ntohs', 'ntohl',
- 'SHUT_RD', 'SHUT_WR', 'SHUT_RDWR',
- ]
-
AF_UNSPEC = 0
AF_INET = 2
AF_INET6 = 23
@@ -165,6 +154,8 @@
SOCK_SEQPACKET = 5 # not supported
SOL_SOCKET = 0xFFFF
+IPPROTO_TCP = 6
+IPPROTO_UDP = 17
SO_BROADCAST = 1
SO_KEEPALIVE = 2
@@ -197,6 +188,28 @@
SO_TYPE = -1024
SO_USELOOPBACK = -2048
+__all__ = ['AF_UNSPEC', 'AF_INET', 'AF_INET6', 'AI_PASSIVE', 'SOCK_DGRAM',
+ 'SOCK_RAW', 'SOCK_RDM', 'SOCK_SEQPACKET', 'SOCK_STREAM', 'SOL_SOCKET',
+ 'SO_BROADCAST', 'SO_ERROR', 'SO_KEEPALIVE', 'SO_LINGER', 'SO_OOBINLINE',
+ 'SO_RCVBUF', 'SO_REUSEADDR', 'SO_SNDBUF', 'SO_TIMEOUT', 'TCP_NODELAY',
+ 'INADDR_ANY', 'INADDR_BROADCAST', 'IPPROTO_TCP', 'IPPROTO_UDP',
+ 'SocketType', 'error', 'herror', 'gaierror', 'timeout',
+ 'getfqdn', 'gethostbyaddr', 'gethostbyname', 'gethostname',
+ 'socket', 'getaddrinfo', 'getdefaulttimeout', 'setdefaulttimeout',
+ 'has_ipv6', 'htons', 'htonl', 'ntohs', 'ntohl',
+ 'SHUT_RD', 'SHUT_WR', 'SHUT_RDWR',
+ ]
+
+def _constant_to_name(const_value):
+ sock_module = sys.modules['socket']
+ try:
+ for name in dir(sock_module):
+ if getattr(sock_module, name) is const_value:
+ return name
+ return "Unknown"
+ finally:
+ sock_module = None
+
class _nio_impl:
timeout = None
@@ -216,9 +229,9 @@
self._timeout_millis = int(timeout*1000)
self.jsocket.setSoTimeout(self._timeout_millis)
- def getsockopt(self, option):
- if self.options.has_key(option):
- result = getattr(self.jsocket, "get%s" % self.options[option])()
+ def getsockopt(self, level, option):
+ if self.options.has_key( (level, option) ):
+ result = getattr(self.jsocket, "get%s" % self.options[ (level, option) ])()
if option == SO_LINGER:
if result == -1:
enabled, linger_time = 0, 0
@@ -227,17 +240,17 @@
return struct.pack('ii', enabled, linger_time)
return result
else:
- raise error(errno.ENOPROTOOPT, "Option not supported on socket(%s): %d" % (str(self.jsocket), option))
+ raise error(errno.ENOPROTOOPT, "Socket option '%s' (level '%s') not supported on socket(%s)" % (_constant_to_name(option), _constant_to_name(level), str(self.jsocket)))
- def setsockopt(self, option, value):
- if self.options.has_key(option):
+ def setsockopt(self, level, option, value):
+ if self.options.has_key( (level, option) ):
if option == SO_LINGER:
values = struct.unpack('ii', value)
self.jsocket.setSoLinger(*values)
else:
- getattr(self.jsocket, "set%s" % self.options[option])(value)
+ getattr(self.jsocket, "set%s" % self.options[ (level, option) ])(value)
else:
- raise error(errno.ENOPROTOOPT, "Option not supported on socket(%s): %d" % (str(self.jsocket), option))
+ raise error(errno.ENOPROTOOPT, "Socket option '%s' (level '%s') not supported on socket(%s)" % (_constant_to_name(option), _constant_to_name(level), str(self.jsocket)))
def close(self):
self.jsocket.close()
@@ -250,14 +263,14 @@
class _client_socket_impl(_nio_impl):
options = {
- SO_KEEPALIVE: 'KeepAlive',
- SO_LINGER: 'SoLinger',
- SO_OOBINLINE: 'OOBInline',
- SO_RCVBUF: 'ReceiveBufferSize',
- SO_REUSEADDR: 'ReuseAddress',
- SO_SNDBUF: 'SendBufferSize',
- SO_TIMEOUT: 'SoTimeout',
- TCP_NODELAY: 'TcpNoDelay',
+ (SOL_SOCKET, SO_KEEPALIVE): 'KeepAlive',
+ (SOL_SOCKET, SO_LINGER): 'SoLinger',
+ (SOL_SOCKET, SO_OOBINLINE): 'OOBInline',
+ (SOL_SOCKET, SO_RCVBUF): 'ReceiveBufferSize',
+ (SOL_SOCKET, SO_REUSEADDR): 'ReuseAddress',
+ (SOL_SOCKET, SO_SNDBUF): 'SendBufferSize',
+ (SOL_SOCKET, SO_TIMEOUT): 'SoTimeout',
+ (IPPROTO_TCP, TCP_NODELAY): 'TcpNoDelay',
}
def __init__(self, socket=None):
@@ -325,9 +338,9 @@
class _server_socket_impl(_nio_impl):
options = {
- SO_RCVBUF: 'ReceiveBufferSize',
- SO_REUSEADDR: 'ReuseAddress',
- SO_TIMEOUT: 'SoTimeout',
+ (SOL_SOCKET, SO_RCVBUF): 'ReceiveBufferSize',
+ (SOL_SOCKET, SO_REUSEADDR): 'ReuseAddress',
+ (SOL_SOCKET, SO_TIMEOUT): 'SoTimeout',
}
def __init__(self, host, port, backlog, reuse_addr):
@@ -353,20 +366,20 @@
return _client_socket_impl(new_cli_sock)
def shutdown(self, how):
- # This is no-op on java, for server sockets.
- # What the user wants to achieve is achieved by calling close() on
- # java/jython. But we can't call that here because that would then
- # later cause the user explicit close() call to fail
- pass
-
+ # This is no-op on java, for server sockets.
+ # What the user wants to achieve is achieved by calling close() on
+ # java/jython. But we can't call that here because that would then
+ # later cause the user explicit close() call to fail
+ pass
+
class _datagram_socket_impl(_nio_impl):
options = {
- SO_BROADCAST: 'Broadcast',
- SO_RCVBUF: 'ReceiveBufferSize',
- SO_REUSEADDR: 'ReuseAddress',
- SO_SNDBUF: 'SendBufferSize',
- SO_TIMEOUT: 'SoTimeout',
+ (SOL_SOCKET, SO_BROADCAST): 'Broadcast',
+ (SOL_SOCKET, SO_RCVBUF): 'ReceiveBufferSize',
+ (SOL_SOCKET, SO_REUSEADDR): 'ReuseAddress',
+ (SOL_SOCKET, SO_SNDBUF): 'SendBufferSize',
+ (SOL_SOCKET, SO_TIMEOUT): 'SoTimeout',
}
def __init__(self, port=None, address=None, reuse_addr=0):
@@ -394,12 +407,12 @@
self.jchannel.disconnect()
def shutdown(self, how):
- # This is no-op on java, for datagram sockets.
- # What the user wants to achieve is achieved by calling close() on
- # java/jython. But we can't call that here because that would then
- # later cause the user explicit close() call to fail
- pass
-
+ # This is no-op on java, for datagram sockets.
+ # What the user wants to achieve is achieved by calling close() on
+ # java/jython. But we can't call that here because that would then
+ # later cause the user explicit close() call to fail
+ pass
+
def _do_send_net(self, byte_array, socket_address, flags):
# Need two separate implementations because the java.nio APIs do not support timeouts
num_bytes = len(byte_array)
@@ -472,10 +485,11 @@
else:
return self._do_receive_nio(0, num_bytes, flags)
+# For now, we DO NOT have complete IPV6 support.
+has_ipv6 = False
+
# Name and address functions
-has_ipv6 = 1
-
def _gethostbyaddr(name):
# This is as close as I can get; at least the types are correct...
addresses = java.net.InetAddress.getAllByName(gethostbyname(name))
@@ -532,13 +546,16 @@
# Same situation as above
raise NotImplementedError("getprotobyname not yet supported on jython.")
-def _realsocket(family = AF_INET, type = SOCK_STREAM, flags=0):
- assert family == AF_INET
- assert type in (SOCK_DGRAM, SOCK_STREAM)
- assert flags == 0
+def _realsocket(family = AF_INET, type = SOCK_STREAM, protocol=0):
+ assert family == AF_INET, "Only AF_INET sockets are currently supported on jython"
+ assert type in (SOCK_DGRAM, SOCK_STREAM), "Only SOCK_STREAM and SOCK_DGRAM sockets are currently supported on jython"
if type == SOCK_STREAM:
+ if protocol != 0:
+ assert protocol == IPPROTO_TCP, "Only IPPROTO_TCP supported on SOCK_STREAM sockets"
return _tcpsocket()
else:
+ if protocol != 0:
+ assert protocol == IPPROTO_UDP, "Only IPPROTO_UDP supported on SOCK_DGRAM sockets"
return _udpsocket()
def getaddrinfo(host, port, family=AF_INET, socktype=None, proto=0, flags=None):
@@ -629,11 +646,11 @@
close_lock = threading.Lock()
def __init__(self):
- self.timeout = _defaulttimeout
- if self.timeout is not None:
- self.mode = MODE_TIMEOUT
+ self.timeout = _defaulttimeout
+ if self.timeout is not None:
+ self.mode = MODE_TIMEOUT
self.pending_options = {
- SO_REUSEADDR: 0,
+ (SOL_SOCKET, SO_REUSEADDR): 0,
}
def gettimeout(self):
@@ -662,34 +679,32 @@
return self.mode == MODE_BLOCKING
def setsockopt(self, level, optname, value):
- if level != SOL_SOCKET: return
try:
if self.sock_impl:
- self.sock_impl.setsockopt(optname, value)
+ self.sock_impl.setsockopt(level, optname, value)
else:
- self.pending_options[optname] = value
+ self.pending_options[ (level, optname) ] = value
except java.lang.Exception, jlx:
raise _map_exception(jlx)
def getsockopt(self, level, optname):
- if level != SOL_SOCKET: return
try:
if self.sock_impl:
- return self.sock_impl.getsockopt(optname)
+ return self.sock_impl.getsockopt(level, optname)
else:
- return self.pending_options.get(optname, None)
+ return self.pending_options.get( (level, optname), None)
except java.lang.Exception, jlx:
raise _map_exception(jlx)
def shutdown(self, how):
assert how in (SHUT_RD, SHUT_WR, SHUT_RDWR)
if not self.sock_impl:
- raise error(errno.ENOTCONN, "Transport endpoint is not connected")
+ raise error(errno.ENOTCONN, "Transport endpoint is not connected")
try:
self.sock_impl.shutdown(how)
except java.lang.Exception, jlx:
raise _map_exception(jlx)
-
+
def close(self):
try:
if self.sock_impl:
@@ -701,9 +716,9 @@
assert self.mode in _permitted_modes
if self.sock_impl:
self.sock_impl.config(self.mode, self.timeout)
- for k in self.pending_options.keys():
- if k != SO_REUSEADDR:
- self.sock_impl.setsockopt(k, self.pending_options[k])
+ for level, optname in self.pending_options.keys():
+ if optname != SO_REUSEADDR:
+ self.sock_impl.setsockopt(level, optname, self.pending_options[ (level, optname) ])
def getchannel(self):
if not self.sock_impl:
@@ -745,7 +760,7 @@
_unpack_address_tuple(addr)
self.local_addr = addr
- def listen(self, backlog=50):
+ def listen(self, backlog):
"This signifies a server socket"
try:
assert not self.sock_impl
@@ -754,7 +769,7 @@
host, port = _unpack_address_tuple(self.local_addr)
else:
host, port = "", 0
- self.sock_impl = _server_socket_impl(host, port, backlog, self.pending_options[SO_REUSEADDR])
+ self.sock_impl = _server_socket_impl(host, port, backlog, self.pending_options[ (SOL_SOCKET, SO_REUSEADDR) ])
self._config()
except java.lang.Exception, jlx:
raise _map_exception(jlx)
@@ -769,7 +784,7 @@
if not new_sock:
raise would_block_error()
cliconn = _tcpsocket()
- cliconn.pending_options[SO_REUSEADDR] = new_sock.jsocket.getReuseAddress()
+ cliconn.pending_options[ (SOL_SOCKET, SO_REUSEADDR) ] = new_sock.jsocket.getReuseAddress()
cliconn.sock_impl = new_sock
cliconn._setup()
return cliconn, new_sock.getpeername()
@@ -789,7 +804,7 @@
self.sock_impl = _client_socket_impl()
if self.local_addr: # Has the socket been bound to a local address?
bind_host, bind_port = _unpack_address_tuple(self.local_addr)
- self.sock_impl.bind(bind_host, bind_port, self.pending_options[SO_REUSEADDR])
+ self.sock_impl.bind(bind_host, bind_port, self.pending_options[ (SOL_SOCKET, SO_REUSEADDR) ])
self._config() # Configure timeouts, etc, now that the socket exists
self.sock_impl.connect(host, port)
except java.lang.Exception, jlx:
@@ -886,7 +901,7 @@
if self.sock_impl:
self.sock_impl.close()
except java.lang.Exception, jlx:
- raise _map_exception(jlx)
+ raise _map_exception(jlx)
class _udpsocket(_nonblocking_api_mixin):
@@ -903,7 +918,7 @@
if host == "":
host = INADDR_ANY
host_address = java.net.InetAddress.getByName(host)
- self.sock_impl = _datagram_socket_impl(port, host_address, self.pending_options[SO_REUSEADDR])
+ self.sock_impl = _datagram_socket_impl(port, host_address, self.pending_options[ (SOL_SOCKET, SO_REUSEADDR) ])
self._config()
except java.lang.Exception, jlx:
raise _map_exception(jlx)
@@ -1003,7 +1018,7 @@
def __del__(self):
self.close()
-
+
_socketmethods = (
'bind', 'connect', 'connect_ex', 'fileno', 'listen',
'getpeername', 'getsockname', 'getsockopt', 'setsockopt',
Modified: branches/Release_2_2maint/jython/Lib/test/test_socket.py
===================================================================
--- branches/Release_2_2maint/jython/Lib/test/test_socket.py 2009-05-20 06:09:43 UTC (rev 6359)
+++ branches/Release_2_2maint/jython/Lib/test/test_socket.py 2009-05-20 15:09:43 UTC (rev 6360)
@@ -21,6 +21,10 @@
MSG = 'Michael Gilfix was here\n'
EIGHT_BIT_MSG = 'Bh\xed Al\xe1in \xd3 Cinn\xe9ide anseo\n'
+import java
+os_name = java.lang.System.getProperty('os.name')
+is_bsd = os_name == 'Mac OS X' or 'BSD' in os_name
+
try:
True
except NameError:
@@ -522,26 +526,27 @@
def setUp(self):
self.test_udp = self.test_tcp_client = self.test_tcp_server = 0
- def _testSetAndGetOption(self, sock, option, values):
+ def _testSetAndGetOption(self, sock, level, option, values):
for expected_value in values:
- sock.setsockopt(socket.SOL_SOCKET, option, expected_value)
- retrieved_value = sock.getsockopt(socket.SOL_SOCKET, option)
+ sock.setsockopt(level, option, expected_value)
+ retrieved_value = sock.getsockopt(level, option)
self.failUnlessEqual(retrieved_value, expected_value, \
- "Retrieved option(%s) value %s != %s(value set)" % (option, retrieved_value, expected_value))
+ "Retrieved option(%s, %s) value %s != %s(value set)" % (level, option, retrieved_value, expected_value))
- def _testUDPOption(self, option, values):
+ def _testUDPOption(self, level, option, values):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- self._testSetAndGetOption(sock, option, values)
+ self._testSetAndGetOption(sock, level, option, values)
# now bind the socket i.e. cause the implementation socket to be created
sock.bind( (HOST, PORT) )
- self.failUnlessEqual(sock.getsockopt(socket.SOL_SOCKET, option), values[-1], \
- "Option value '%s'='%s' did not propagate to implementation socket" % (option, values[-1]) )
- self._testSetAndGetOption(sock, option, values)
+ self.failUnlessEqual(sock.getsockopt(level, option), values[-1], \
+ "Option value '(%s, %s)'='%s' did not propagate to implementation socket" % (level, option, values[-1]) )
+ self._testSetAndGetOption(sock, level, option, values)
finally:
sock.close()
- def _testTCPClientOption(self, option, values):
+ def _testTCPClientOption(self, level, option, values):
+ sock = None
try:
# First listen on a server socket, so that the connection won't be refused.
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
@@ -549,42 +554,51 @@
server_sock.listen(50)
# Now do the tests
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self._testSetAndGetOption(sock, option, values)
+ self._testSetAndGetOption(sock, level, option, values)
# now connect the socket i.e. cause the implementation socket to be created
# First bind, so that the SO_REUSEADDR setting propagates
sock.bind( (HOST, PORT+1) )
sock.connect( (HOST, PORT) )
- self.failUnlessEqual(sock.getsockopt(socket.SOL_SOCKET, option), values[-1], \
- "Option value '%s'='%s' did not propagate to implementation socket" % (option, values[-1]))
- self._testSetAndGetOption(sock, option, values)
+ msg = "Option value '%s'='%s' did not propagate to implementation socket" % (option, values[-1])
+ if is_bsd and option in (socket.SO_RCVBUF, socket.SO_SNDBUF):
+ # XXX: there's no guarantee that bufsize will be the
+ # exact setsockopt value, particularly after
+ # establishing a connection. seems it will be *at least*
+ # the values we test (which are rather small) on
+ # BSDs. may need to relax this on other platforms also
+ self.assert_(sock.getsockopt(level, option) >= values[-1], msg)
+ else:
+ self.failUnlessEqual(sock.getsockopt(level, option), values[-1], msg)
+ self._testSetAndGetOption(sock, level, option, values)
finally:
server_sock.close()
- sock.close()
+ if sock:
+ sock.close()
- def _testTCPServerOption(self, option, values):
+ def _testTCPServerOption(self, level, option, values):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self._testSetAndGetOption(sock, option, values)
+ self._testSetAndGetOption(sock, level, option, values)
# now bind and listen on the socket i.e. cause the implementation socket to be created
sock.bind( (HOST, PORT) )
sock.listen(50)
- self.failUnlessEqual(sock.getsockopt(socket.SOL_SOCKET, option), values[-1], \
- "Option value '%s'='%s' did not propagate to implementation socket" % (option, values[-1]))
- self._testSetAndGetOption(sock, option, values)
+ self.failUnlessEqual(sock.getsockopt(level, option), values[-1], \
+ "Option value '(%s,%s)'='%s' did not propagate to implementation socket" % (level, option, values[-1]))
+ self._testSetAndGetOption(sock, level, option, values)
finally:
sock.close()
- def _testOption(self, option, values):
+ def _testOption(self, level, option, values):
for flag, func in [
(self.test_udp, self._testUDPOption),
(self.test_tcp_server, self._testTCPServerOption),
(self.test_tcp_client, self._testTCPClientOption),
]:
if flag:
- func(option, values)
+ func(level, option, values)
else:
try:
- func(option, values)
+ func(level, option, values)
except socket.error, se:
self.failUnlessEqual(se[0], errno.ENOPROTOOPT, "Wrong errno from unsupported option exception: %d" % se[0])
except Exception, x:
@@ -596,48 +610,48 @@
def testSO_BROADCAST(self):
self.test_udp = 1
- self._testOption(socket.SO_BROADCAST, [0, 1])
+ self._testOption(socket.SOL_SOCKET, socket.SO_BROADCAST, [0, 1])
def testSO_KEEPALIVE(self):
self.test_tcp_client = 1
- self._testOption(socket.SO_KEEPALIVE, [0, 1])
+ self._testOption(socket.SOL_SOCKET, socket.SO_KEEPALIVE, [0, 1])
def testSO_LINGER(self):
self.test_tcp_client = 1
off = struct.pack('ii', 0, 0)
on_2_seconds = struct.pack('ii', 1, 2)
- self._testOption(socket.SO_LINGER, [off, on_2_seconds])
+ self._testOption(socket.SOL_SOCKET, socket.SO_LINGER, [off, on_2_seconds])
def testSO_OOBINLINE(self):
self.test_tcp_client = 1
- self._testOption(socket.SO_OOBINLINE, [0, 1])
+ self._testOption(socket.SOL_SOCKET, socket.SO_OOBINLINE, [0, 1])
def testSO_RCVBUF(self):
self.test_udp = 1
self.test_tcp_client = 1
self.test_tcp_server = 1
- self._testOption(socket.SO_RCVBUF, [1024, 4096, 16384])
+ self._testOption(socket.SOL_SOCKET, socket.SO_RCVBUF, [1024, 4096, 16384])
def testSO_REUSEADDR(self):
self.test_udp = 1
self.test_tcp_client = 1
self.test_tcp_server = 1
- self._testOption(socket.SO_REUSEADDR, [0, 1])
+ self._testOption(socket.SOL_SOCKET, socket.SO_REUSEADDR, [0, 1])
def testSO_SNDBUF(self):
self.test_udp = 1
self.test_tcp_client = 1
- self._testOption(socket.SO_SNDBUF, [1024, 4096, 16384])
+ self._testOption(socket.SOL_SOCKET, socket.SO_SNDBUF, [1024, 4096, 16384])
def testSO_TIMEOUT(self):
self.test_udp = 1
self.test_tcp_client = 1
self.test_tcp_server = 1
- self._testOption(socket.SO_TIMEOUT, [0, 1, 1000])
+ self._testOption(socket.SOL_SOCKET, socket.SO_TIMEOUT, [0, 1, 1000])
def testTCP_NODELAY(self):
self.test_tcp_client = 1
- self._testOption(socket.TCP_NODELAY, [0, 1])
+ self._testOption(socket.IPPROTO_TCP, socket.TCP_NODELAY, [0, 1])
class TestUnsupportedOptions(TestSocketOptions):
@@ -795,31 +809,31 @@
self.serv_conn.send(MSG)
self.serv_conn.send('and ' + MSG)
-class UDPBindTest(unittest.TestCase):
-
+class UDPBindTest(unittest.TestCase):
+
HOST = HOST
PORT = PORT
- def setUp(self):
+ def setUp(self):
self.sock = socket.socket(socket.AF_INET,socket.SOCK_DGRAM)
-
- def testBindSpecific(self):
+
+ def testBindSpecific(self):
self.sock.bind( (self.HOST, self.PORT) ) # Use a specific port
- actual_port = self.sock.getsockname()[1]
- self.failUnless(actual_port == self.PORT,
- "Binding to specific port number should have returned same number: %d != %d" % (actual_port, self.PORT))
+ actual_port = self.sock.getsockname()[1]
+ self.failUnless(actual_port == self.PORT,
+ "Binding to specific port number should have returned same number: %d != %d" % (actual_port, self.PORT))
- def testBindEphemeral(self):
+ def testBindEphemeral(self):
self.sock.bind( (self.HOST, 0) ) # let system choose a free port
- self.failUnless(self.sock.getsockname()[1] != 0, "Binding to port zero should have allocated an ephemeral port number")
+ self.failUnless(self.sock.getsockname()[1] != 0, "Binding to port zero should have allocated an ephemeral port number")
- def testShutdown(self):
- self.sock.bind( (self.HOST, self.PORT) )
- self.sock.shutdown(socket.SHUT_RDWR)
+ def testShutdown(self):
+ self.sock.bind( (self.HOST, self.PORT) )
+ self.sock.shutdown(socket.SHUT_RDWR)
- def tearDown(self):
- self.sock.close()
-
+ def tearDown(self):
+ self.sock.close()
+
class BasicUDPTest(ThreadedUDPSocketTest):
def __init__(self, methodName='runTest'):
@@ -1327,7 +1341,7 @@
used, but if it is on your network this failure is bogus.''' % host)
def testConnectDefaultTimeout(self):
- _saved_timeout = socket.getdefaulttimeout()
+ _saved_timeout = socket.getdefaulttimeout()
socket.setdefaulttimeout(0.1)
cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
host = '192.168.192.168'
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|
