Menu
▾
▴
[Jython-checkins] SF.net SVN: jython: [3319] trunk/jython/Lib
|
From: <am...@us...> - 2007-07-16 19:13:27
|
Revision: 3319
http://svn.sourceforge.net/jython/?rev=3319&view=rev
Author: amak
Date: 2007-07-16 12:13:26 -0700 (Mon, 16 Jul 2007)
Log Message:
-----------
. Fixed a bug with FileWrappers, whereby closing the FileWrapper incorrectly caused closing of the underlying socket.
. Wrapped several more methods in try .. except for exception mapping
. Removed unnecessary lambdas from exception mapping table
. Completed the exception mapping table with all java.net and java.nio exceptions
. Removed errno module simulation: was unnecessary (jython 2.1 users can easily create their own errno module)
. Added constants SHUT_RD, SHUT_WR and SHUT_RDWR, as used by shutdown()
. Added proper socket shutdown support, as opposed to the old method of closing the [In|Out]putStreams
. Removed some comments and commented-out code
. Did some basic refactoring of UDP connect methods, to reduce code duplication
. Added two more unit tests for closing sockets and FileWrappers
. Added more unit tests for exception mapping
Modified Paths:
--------------
trunk/jython/Lib/socket.py
trunk/jython/Lib/test/test_socket.py
Modified: trunk/jython/Lib/socket.py
===================================================================
--- trunk/jython/Lib/socket.py 2007-07-16 02:44:37 UTC (rev 3318)
+++ trunk/jython/Lib/socket.py 2007-07-16 19:13:26 UTC (rev 3319)
@@ -16,50 +16,67 @@
_defaulttimeout = None
+import errno
+import jarray
+import string
+import sys
import threading
import time
import types
-import jarray
-import string
-import sys
+# Java.io classes
import java.io.BufferedInputStream
import java.io.BufferedOutputStream
+# Java.io exceptions
import java.io.InterruptedIOException
+import java.io.IOException
+
+# Java.lang classes
+import java.lang.String
+# Java.lang exceptions
import java.lang.Exception
-import java.lang.String
-import java.net.BindException
-import java.net.ConnectException
+
+# Java.net classes
import java.net.DatagramPacket
import java.net.InetAddress
import java.net.InetSocketAddress
import java.net.Socket
+# Java.net exceptions
+import java.net.BindException
+import java.net.ConnectException
+import java.net.NoRouteToHostException
+import java.net.PortUnreachableException
+import java.net.ProtocolException
+import java.net.SocketException
import java.net.SocketTimeoutException
import java.net.UnknownHostException
+
+# Java.nio classes
import java.nio.ByteBuffer
import java.nio.channels.DatagramChannel
-import java.nio.channels.IllegalBlockingModeException
import java.nio.channels.ServerSocketChannel
import java.nio.channels.SocketChannel
+# Java.nio exceptions
+import java.nio.channels.AlreadyConnectedException
+import java.nio.channels.AsynchronousCloseException
+import java.nio.channels.CancelledKeyException
+import java.nio.channels.ClosedByInterruptException
+import java.nio.channels.ClosedChannelException
+import java.nio.channels.ClosedSelectorException
+import java.nio.channels.ConnectionPendingException
+import java.nio.channels.IllegalBlockingModeException
+import java.nio.channels.IllegalSelectorException
+import java.nio.channels.NoConnectionPendingException
+import java.nio.channels.NonReadableChannelException
+import java.nio.channels.NonWritableChannelException
+import java.nio.channels.NotYetBoundException
+import java.nio.channels.NotYetConnectedException
import java.nio.channels.UnresolvedAddressException
+import java.nio.channels.UnsupportedAddressTypeException
+
import javax.net.ssl.SSLSocketFactory
import org.python.core.PyFile
-try:
- import errno
- ERRNO_EWOULDBLOCK = errno.EWOULDBLOCK
- ERRNO_EACCES = errno.EACCES
- ERRNO_ECONNREFUSED = errno.ECONNREFUSED
- ERRNO_EINPROGRESS = errno.EINPROGRESS
- ERRNO_EGETADDRINFOFAILED = errno.EGETADDRINFOFAILED
-except ImportError:
- # Support jython 2.1
- ERRNO_EWOULDBLOCK = 11
- ERRNO_EACCES = 13
- ERRNO_ECONNREFUSED = 111
- ERRNO_EINPROGRESS = 115
- ERRNO_EGETADDRINFOFAILED = 20001
-
class error(Exception): pass
class herror(error): pass
class gaierror(error): pass
@@ -71,23 +88,48 @@
# (<javaexception>, <circumstance>) : lambda: <code that raises the python equivalent>
-(java.io.InterruptedIOException, ALL) : lambda exc: timeout('timed out'),
-(java.net.BindException, ALL) : lambda exc: error(ERRNO_EACCES, 'Permission denied'),
-(java.net.ConnectException, ALL) : lambda exc: error(ERRNO_ECONNREFUSED, 'Connection refused'),
-(java.net.SocketTimeoutException, ALL) : lambda exc: timeout('timed out'),
-(java.net.UnknownHostException, ALL) : lambda exc: gaierror(ERRNO_EGETADDRINFOFAILED, 'getaddrinfo failed'),
-(java.nio.channels.UnresolvedAddressException, ALL) : lambda exc: gaierror(ERRNO_EGETADDRINFOFAILED, 'getaddrinfo failed'),
+(java.io.IOException, ALL) : error(errno.ECONNRESET, 'Software caused connection abort'),
+(java.io.InterruptedIOException, ALL) : timeout('timed out'),
+(java.net.BindException, ALL) : error(errno.EADDRINUSE, 'Address already in use'),
+(java.net.ConnectException, ALL) : error(errno.ECONNREFUSED, 'Connection refused'),
+(java.net.NoRouteToHostException, ALL) : error(-1, 'Unmapped exception: java.net.NoRouteToHostException'),
+(java.net.PortUnreachableException, ALL) : error(-1, 'Unmapped exception: java.net.PortUnreachableException'),
+(java.net.ProtocolException, ALL) : error(-1, 'Unmapped exception: java.net.ProtocolException'),
+(java.net.SocketException, ALL) : error(-1, 'Unmapped exception: java.net.SocketException'),
+(java.net.SocketTimeoutException, ALL) : timeout('timed out'),
+(java.net.UnknownHostException, ALL) : gaierror(errno.EGETADDRINFOFAILED, 'getaddrinfo failed'),
+
+(java.nio.channels.AlreadyConnectedException, ALL) : error(errno.EISCONN, 'Socket is already connected'),
+(java.nio.channels.AsynchronousCloseException, ALL) : error(-1, 'Unmapped exception: java.nio.AsynchronousCloseException'),
+(java.nio.channels.CancelledKeyException, ALL) : error(-1, 'Unmapped exception: java.nio.CancelledKeyException'),
+(java.nio.channels.ClosedByInterruptException, ALL) : error(-1, 'Unmapped exception: java.nio.ClosedByInterruptException'),
+(java.nio.channels.ClosedChannelException, ALL) : error(errno.EPIPE, 'Socket closed'),
+(java.nio.channels.ClosedSelectorException, ALL) : error(-1, 'Unmapped exception: java.nio.ClosedSelectorException'),
+(java.nio.channels.ConnectionPendingException, ALL) : error(-1, 'Unmapped exception: java.nio.ConnectionPendingException'),
+(java.nio.channels.IllegalBlockingModeException, ALL) : error(-1, 'Unmapped exception: java.nio.IllegalBlockingModeException'),
+(java.nio.channels.IllegalSelectorException, ALL) : error(-1, 'Unmapped exception: java.nio.IllegalSelectorException'),
+(java.nio.channels.NoConnectionPendingException, ALL) : error(-1, 'Unmapped exception: java.nio.NoConnectionPendingException'),
+(java.nio.channels.NonReadableChannelException, ALL) : error(-1, 'Unmapped exception: java.nio.NonReadableChannelException'),
+(java.nio.channels.NonWritableChannelException, ALL) : error(-1, 'Unmapped exception: java.nio.NonWritableChannelException'),
+(java.nio.channels.NotYetBoundException, ALL) : error(-1, 'Unmapped exception: java.nio.NotYetBoundException'),
+(java.nio.channels.NotYetConnectedException, ALL) : error(-1, 'Unmapped exception: java.nio.NotYetConnectedException'),
+(java.nio.channels.UnresolvedAddressException, ALL) : gaierror(errno.EGETADDRINFOFAILED, 'getaddrinfo failed'),
+(java.nio.channels.UnsupportedAddressTypeException, ALL) : error(-1, 'Unmapped exception: java.nio.UnsupportedAddressTypeException'),
+
}
def would_block_error(exc=None):
- return error( (ERRNO_EWOULDBLOCK, 'The socket operation could not complete without blocking') )
+ return error(errno.EWOULDBLOCK, 'The socket operation could not complete without blocking')
def _map_exception(exc, circumstance=ALL):
+# print "Mapping exception: %s" % exc
try:
- return _exception_map[(exc.__class__, circumstance)](exc)
+ mapped_exception = _exception_map[(exc.__class__, circumstance)]
+ mapped_exception.java_exception = exc
+ return mapped_exception
except KeyError:
- return error(-1, 'Unmapped java exception: %s' % exc.toString())
+ return error(-1, 'Unmapped java exception: <%s:%s>' % (exc.toString(), circumstance))
MODE_BLOCKING = 'block'
MODE_NONBLOCKING = 'nonblock'
@@ -95,6 +137,10 @@
_permitted_modes = (MODE_BLOCKING, MODE_NONBLOCKING, MODE_TIMEOUT)
+SHUT_RD = 0
+SHUT_WR = 1
+SHUT_RDWR = 2
+
class _nio_impl:
timeout = None
@@ -152,6 +198,12 @@
# close = close3
# close = close4
+ def shutdownInput(self):
+ self.jsocket.shutdownInput()
+
+ def shutdownOutput(self):
+ self.jsocket.shutdownOutput()
+
def getchannel(self):
return self.jchannel
@@ -184,9 +236,6 @@
def finish_connect(self):
return self.jchannel.finishConnect()
- def close(self):
- _nio_impl.close(self)
-
class _server_socket_impl(_nio_impl):
def __init__(self, host, port, backlog, reuse_addr):
@@ -200,23 +249,17 @@
self.jsocket.bind(bindaddr, backlog)
def accept(self):
- try:
- if self.mode in (MODE_BLOCKING, MODE_NONBLOCKING):
- new_cli_chan = self.jchannel.accept()
- if new_cli_chan != None:
- return _client_socket_impl(new_cli_chan.socket())
- else:
- return None
+ if self.mode in (MODE_BLOCKING, MODE_NONBLOCKING):
+ new_cli_chan = self.jchannel.accept()
+ if new_cli_chan != None:
+ return _client_socket_impl(new_cli_chan.socket())
else:
- # In timeout mode now
- new_cli_sock = self.jsocket.accept()
- return _client_socket_impl(new_cli_sock)
- except java.lang.Exception, jlx:
- raise _map_exception(jlx)
+ return None
+ else:
+ # In timeout mode now
+ new_cli_sock = self.jsocket.accept()
+ return _client_socket_impl(new_cli_sock)
- def close(self):
- _nio_impl.close(self)
-
class _datagram_socket_impl(_nio_impl):
def __init__(self, port=None, address=None, reuse_addr=0):
@@ -248,6 +291,7 @@
'getfqdn', 'gethostbyaddr', 'gethostbyname', 'gethostname',
'socket', 'getaddrinfo', 'getdefaulttimeout', 'setdefaulttimeout',
'has_ipv6', 'htons', 'htonl', 'ntohs', 'ntohl',
+ 'SHUT_RD', 'SHUT_WR', 'SHUT_RDWR',
]
AF_INET = 2
@@ -343,7 +387,7 @@
floatvalue = float(value)
except:
raise TypeError('Socket timeout value must be a number or None')
- if floatvalue < 0:
+ if floatvalue < 0.0:
raise ValueError("Socket timeout value cannot be negative")
if floatvalue < 0.000001:
return 0.0
@@ -396,9 +440,6 @@
if not self.sock_impl:
return None
return self.sock_impl.getchannel()
-# if hasattr(self.sock_impl, 'getchannel'):
-# return self.sock_impl.getchannel()
-# raise error('Operation not implemented on this JVM')
fileno = getchannel
@@ -421,7 +462,6 @@
local_addr = None
server = 0
file_count = 0
- #reuse_addr = 1
reuse_addr = 0
def bind(self, addr):
@@ -445,11 +485,6 @@
except java.lang.Exception, jlx:
raise _map_exception(jlx)
-#
-# The following has information on a java.lang.NullPointerException problem I'm having
-#
-# http://developer.java.sun.com/developer/bugParade/bugs/4801882.html
-
def accept(self):
"This signifies a server socket"
try:
@@ -495,7 +530,7 @@
if self.sock_impl.finish_connect():
self._setup(self.sock_impl)
return 0
- return ERRNO_EINPROGRESS
+ return errno.EINPROGRESS
def _setup(self, sock):
self.sock_impl = sock
@@ -506,7 +541,7 @@
def recv(self, n):
try:
- if not self.sock_impl: raise error('Socket not open')
+ if not self.sock_impl: raise error(errno.ENOTCONN, 'Socket is not connected')
if self.sock_impl.jchannel.isConnectionPending():
self.sock_impl.jchannel.finishConnect()
data = jarray.zeros(n, 'b')
@@ -525,33 +560,41 @@
return self.recv(n), None
def send(self, s):
- if not self.sock_impl: raise error('Socket not open')
- if self.sock_impl.jchannel.isConnectionPending():
- self.sock_impl.jchannel.finishConnect()
- #n = len(s)
- numwritten = self.sock_impl.write(s)
- return numwritten
+ try:
+ if not self.sock_impl: raise error(errno.ENOTCONN, 'Socket is not connected')
+ if self.sock_impl.jchannel.isConnectionPending():
+ self.sock_impl.jchannel.finishConnect()
+ numwritten = self.sock_impl.write(s)
+ return numwritten
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
sendall = send
def getsockname(self):
- if not self.sock_impl:
- host, port = self.local_addr or ("", 0)
- host = java.net.InetAddress.getByName(host).getHostAddress()
- else:
- if self.server:
- host = self.sock_impl.jsocket.getInetAddress().getHostAddress()
+ try:
+ if not self.sock_impl:
+ host, port = self.local_addr or ("", 0)
+ host = java.net.InetAddress.getByName(host).getHostAddress()
else:
- host = self.sock_impl.jsocket.getLocalAddress().getHostAddress()
- port = self.sock_impl.jsocket.getLocalPort()
- return (host, port)
+ if self.server:
+ host = self.sock_impl.jsocket.getInetAddress().getHostAddress()
+ else:
+ host = self.sock_impl.jsocket.getLocalAddress().getHostAddress()
+ port = self.sock_impl.jsocket.getLocalPort()
+ return (host, port)
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def getpeername(self):
- assert self.sock_impl
- assert not self.server
- host = self.sock_impl.jsocket.getInetAddress().getHostAddress()
- port = self.sock_impl.jsocket.getPort()
- return (host, port)
+ try:
+ assert self.sock_impl
+ assert not self.server
+ host = self.sock_impl.jsocket.getInetAddress().getHostAddress()
+ port = self.sock_impl.jsocket.getPort()
+ return (host, port)
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def setsockopt(self, level, optname, value):
if optname == SO_REUSEADDR:
@@ -578,12 +621,11 @@
class FileWrapper:
def __init__(self, socket, file):
- self.socket = socket
- self.sock = socket.sock_impl
+ self.socket = socket
self.istream = socket.istream
self.ostream = socket.ostream
- self.file = file
+ self.file = file
self.read = file.read
self.readline = file.readline
self.readlines = file.readlines
@@ -597,51 +639,62 @@
self.socket.file_count += 1
def close(self):
- if self.file.closed:
+ if self.closed:
# Already closed
return
self.socket.file_count -= 1
- self.file.close()
- self.closed = self.file.closed
+ # AMAK: 20070715: Cannot close the PyFile, because closing
+ # it causes the InputStream and OutputStream to be closed.
+ # This in turn causes the underlying socket to be closed.
+ # This was always true for java.net sockets
+ # And continues to be true for java.nio sockets
+ # http://bugs.sun.com/bugdatabase/view_bug.do?bug_id=4717638
+# self.file.close()
+ istream = self.istream
+ ostream = self.ostream
+ self.istream = None
+ self.ostream = None
+# self.closed = self.file.closed
+ self.closed = 1
- if self.socket.file_count == 0 and self.socket.sock_impl == 0:
+ if self.socket.file_count == 0 and self.socket.sock_impl is None:
# This is the last file Only close the socket and streams
# if there are no outstanding files left.
- if self.sock:
- self.sock.close()
- if self.istream:
- self.istream.close()
- if self.ostream:
- self.ostream.close()
+ istream.close()
+ ostream.close()
def shutdown(self, how):
- assert how in (0, 1, 2)
+ assert how in (SHUT_RD, SHUT_WR, SHUT_RDWR)
assert self.sock_impl
- if how in (0, 2):
+ if how in (SHUT_RD, SHUT_RDWR):
+ self.sock_impl.shutdownInput()
+ if how in (SHUT_WR, SHUT_RDWR):
+ self.sock_impl.shutdownOutput()
+
+ def close(self):
+ try:
+ if not self.sock_impl:
+ return
+ sock_impl = self.sock_impl
+ istream = self.istream
+ ostream = self.ostream
+ self.sock_impl = None
self.istream = None
- if how in (1, 2):
self.ostream = None
+ # Only close the socket and streams if there are no
+ # outstanding files left.
+ if self.file_count == 0:
+ if istream:
+ istream.close()
+ if ostream:
+ ostream.close()
+ if sock_impl:
+ sock_impl.close()
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
+
- def close(self):
- if not self.sock_impl:
- return
- sock = self.sock_impl
- istream = self.istream
- ostream = self.ostream
- self.sock_impl = 0
- self.istream = 0
- self.ostream = 0
- # Only close the socket and streams if there are no
- # outstanding files left.
- if self.file_count == 0:
- if istream:
- istream.close()
- if ostream:
- ostream.close()
- if sock:
- sock.close()
-
class _udpsocket(_nonblocking_api_mixin):
def __init__(self):
@@ -650,50 +703,56 @@
self.reuse_addr = 0
def bind(self, addr):
- assert not self.sock_impl
- host, port = _unpack_address_tuple(addr)
- host_address = java.net.InetAddress.getByName(host)
- self.sock_impl = _datagram_socket_impl(port, host_address, reuse_addr = self.reuse_addr)
- self._config()
+ try:
+ assert not self.sock_impl
+ host, port = _unpack_address_tuple(addr)
+ host_address = java.net.InetAddress.getByName(host)
+ self.sock_impl = _datagram_socket_impl(port, host_address, reuse_addr = self.reuse_addr)
+ self._config()
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
+ def _do_connect(self, addr):
+ try:
+ host, port = _unpack_address_tuple(addr)
+ assert not self.addr
+ self.addr = addr
+ if not self.sock_impl:
+ self.sock_impl = _datagram_socket_impl()
+ self._config()
+ self.sock_impl.connect(host, port)
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
+
def connect(self, addr):
- host, port = _unpack_address_tuple(addr)
- assert not self.addr
- if not self.sock_impl:
- self.sock_impl = _datagram_socket_impl()
- self._config()
- self.sock_impl.connect(host, port)
- self.addr = addr # convert host to InetAddress instance?
+ self._do_connect(addr)
def connect_ex(self, addr):
- host, port = _unpack_address_tuple(addr)
- assert not self.addr
- self.addr = addr
- if not self.sock_impl:
- self.sock_impl = _datagram_socket_impl()
- self._config()
- self.sock_impl.connect(host, port)
- if self.sock_impl.finish_connect():
- return 0
- return ERRNO_EINPROGRESS
+ self._do_connect(addr)
+ if self.sock_impl.finish_connect():
+ return 0
+ return errno.EINPROGRESS
def sendto(self, data, p1, p2=None):
- if not p2:
- flags, addr = 0, p1
- else:
- flags, addr = 0, p2
- n = len(data)
- if not self.sock_impl:
- self.sock_impl = _datagram_socket_impl()
- host, port = addr
- bytes = java.lang.String(data).getBytes('iso-8859-1')
- a = java.net.InetAddress.getByName(host)
- packet = java.net.DatagramPacket(bytes, n, a, port)
- self.sock_impl.send(packet)
- return n
+ try:
+ if not p2:
+ flags, addr = 0, p1
+ else:
+ flags, addr = 0, p2
+ n = len(data)
+ if not self.sock_impl:
+ self.sock_impl = _datagram_socket_impl()
+ host, port = addr
+ bytes = java.lang.String(data).getBytes('iso-8859-1')
+ a = java.net.InetAddress.getByName(host)
+ packet = java.net.DatagramPacket(bytes, n, a, port)
+ self.sock_impl.send(packet)
+ return n
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def send(self, data):
- assert self.addr
+ if not self.addr: raise error(errno.ENOTCONN, "Socket is not connected")
return self.sendto(data, self.addr)
def recvfrom(self, n):
@@ -727,37 +786,52 @@
raise _map_exception(jlx)
def getsockname(self):
- assert self.sock_impl
- host = self.sock_impl.jsocket.getLocalAddress().getHostName()
- port = self.sock_impl.jsocket.getLocalPort()
- return (host, port)
+ try:
+ assert self.sock_impl
+ host = self.sock_impl.jsocket.getLocalAddress().getHostName()
+ port = self.sock_impl.jsocket.getLocalPort()
+ return (host, port)
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def getpeername(self):
- assert self.sock
- host = self.sock_impl.jsocket.getInetAddress().getHostName()
- port = self.sock_impl.jsocket.getPort()
- return (host, port)
+ try:
+ assert self.sock
+ host = self.sock_impl.jsocket.getInetAddress().getHostName()
+ port = self.sock_impl.jsocket.getPort()
+ return (host, port)
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def __del__(self):
self.close()
def close(self):
- if not self.sock_impl:
- return
- sock = self.sock_impl
- self.sock_impl = None
- sock.close()
+ try:
+ if not self.sock_impl:
+ return
+ sock = self.sock_impl
+ self.sock_impl = None
+ sock.close()
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def setsockopt(self, level, optname, value):
- if optname == SO_REUSEADDR:
- self.reuse_addr = value
-# self.sock._setreuseaddress(value)
+ try:
+ if optname == SO_REUSEADDR:
+ self.reuse_addr = value
+# self.sock._setreuseaddress(value)
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
def getsockopt(self, level, optname):
- if optname == SO_REUSEADDR:
- return self.sock_impl._getreuseaddress()
- else:
- return None
+ try:
+ if optname == SO_REUSEADDR:
+ return self.sock_impl._getreuseaddress()
+ else:
+ return None
+ except java.lang.Exception, jlx:
+ raise _map_exception(jlx)
SocketType = _tcpsocket
SocketTypes = [_tcpsocket, _udpsocket]
Modified: trunk/jython/Lib/test/test_socket.py
===================================================================
--- trunk/jython/Lib/test/test_socket.py 2007-07-16 02:44:37 UTC (rev 3318)
+++ trunk/jython/Lib/test/test_socket.py 2007-07-16 19:13:26 UTC (rev 3319)
@@ -1,1059 +1,1121 @@
-from __future__ import nested_scopes
-
-"""
-AMAK: 20050515: This module is the test_socket.py from cpython 2.4, ported to jython.
-"""
-
-import unittest
-#from test import test_support
-
-import errno
-import socket
-import select
-import time
-import thread, threading
-import Queue
-import sys
-from weakref import proxy
-
-PORT = 50007
-HOST = 'localhost'
-MSG = 'Michael Gilfix was here\n'
-EIGHT_BIT_MSG = 'Bh\xed Al\xe1in \xd3 Cinn\xe9ide anseo\n'
-
-try:
- True
-except NameError:
- True, False = 1, 0
-
-class SocketTCPTest(unittest.TestCase):
-
- def setUp(self):
- self.serv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- self.serv.bind((HOST, PORT))
- self.serv.listen(1)
-
- def tearDown(self):
- self.serv.close()
- self.serv = None
-
-class SocketUDPTest(unittest.TestCase):
-
- def setUp(self):
- self.serv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- self.serv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- self.serv.bind((HOST, PORT))
-
- def tearDown(self):
- self.serv.close()
- self.serv = None
-
-class ThreadableTest:
- """Threadable Test class
-
- The ThreadableTest class makes it easy to create a threaded
- client/server pair from an existing unit test. To create a
- new threaded class from an existing unit test, use multiple
- inheritance:
-
- class NewClass (OldClass, ThreadableTest):
- pass
-
- This class defines two new fixture functions with obvious
- purposes for overriding:
-
- clientSetUp ()
- clientTearDown ()
-
- Any new test functions within the class must then define
- tests in pairs, where the test name is preceeded with a
- '_' to indicate the client portion of the test. Ex:
-
- def testFoo(self):
- # Server portion
-
- def _testFoo(self):
- # Client portion
-
- Any exceptions raised by the clients during their tests
- are caught and transferred to the main thread to alert
- the testing framework.
-
- Note, the server setup function cannot call any blocking
- functions that rely on the client thread during setup,
- unless serverExplicityReady() is called just before
- the blocking call (such as in setting up a client/server
- connection and performing the accept() in setUp().
- """
-
- def __init__(self):
- # Swap the true setup function
- self.__setUp = self.setUp
- self.__tearDown = self.tearDown
- self.setUp = self._setUp
- self.tearDown = self._tearDown
-
- def serverExplicitReady(self):
- """This method allows the server to explicitly indicate that
- it wants the client thread to proceed. This is useful if the
- server is about to execute a blocking routine that is
- dependent upon the client thread during its setup routine."""
- self.server_ready.set()
-
- def _setUp(self):
- self.server_ready = threading.Event()
- self.client_ready = threading.Event()
- self.done = threading.Event()
- self.queue = Queue.Queue(1)
-
- # Do some munging to start the client test.
- methodname = self.id()
- i = methodname.rfind('.')
- methodname = methodname[i+1:]
- self.test_method_name = methodname
- test_method = getattr(self, '_' + methodname)
- self.client_thread = thread.start_new_thread(
- self.clientRun, (test_method,))
-
- self.__setUp()
- if not self.server_ready.isSet():
- self.server_ready.set()
- self.client_ready.wait()
-
- def _tearDown(self):
- self.__tearDown()
- self.done.wait()
-
- if not self.queue.empty():
- msg = self.queue.get()
- self.fail(msg)
-
- def clientRun(self, test_func):
- self.server_ready.wait()
- self.client_ready.set()
- self.clientSetUp()
- if not callable(test_func):
- raise TypeError, "test_func must be a callable function"
- try:
- test_func()
- except Exception, strerror:
- self.queue.put(strerror)
- self.clientTearDown()
-
- def clientSetUp(self):
- raise NotImplementedError, "clientSetUp must be implemented."
-
- def clientTearDown(self):
- self.done.set()
- if sys.platform[:4] != 'java':
- # This causes the whole process to exit on jython
- # Probably related to problems with daemon status of threads
- thread.exit()
-
-class ThreadedTCPSocketTest(SocketTCPTest, ThreadableTest):
-
- def __init__(self, methodName='runTest'):
- SocketTCPTest.__init__(self, methodName=methodName)
- ThreadableTest.__init__(self)
-
- def clientSetUp(self):
- self.cli = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-
- def clientTearDown(self):
- self.cli.close()
- self.cli = None
- ThreadableTest.clientTearDown(self)
-
-class ThreadedUDPSocketTest(SocketUDPTest, ThreadableTest):
-
- def __init__(self, methodName='runTest'):
- SocketUDPTest.__init__(self, methodName=methodName)
- ThreadableTest.__init__(self)
-
- def clientSetUp(self):
- self.cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-
-class SocketConnectedTest(ThreadedTCPSocketTest):
-
- def __init__(self, methodName='runTest'):
- ThreadedTCPSocketTest.__init__(self, methodName=methodName)
-
- def setUp(self):
- ThreadedTCPSocketTest.setUp(self)
- # Indicate explicitly we're ready for the client thread to
- # proceed and then perform the blocking call to accept
- self.serverExplicitReady()
- conn, addr = self.serv.accept()
- self.cli_conn = conn
-
- def tearDown(self):
- self.cli_conn.close()
- self.cli_conn = None
- ThreadedTCPSocketTest.tearDown(self)
-
- def clientSetUp(self):
- ThreadedTCPSocketTest.clientSetUp(self)
- self.cli.connect((HOST, PORT))
- self.serv_conn = self.cli
-
- def clientTearDown(self):
- self.serv_conn.close()
- self.serv_conn = None
- ThreadedTCPSocketTest.clientTearDown(self)
-
-class SocketPairTest(unittest.TestCase, ThreadableTest):
-
- def __init__(self, methodName='runTest'):
- unittest.TestCase.__init__(self, methodName=methodName)
- ThreadableTest.__init__(self)
-
- def setUp(self):
- self.serv, self.cli = socket.socketpair()
-
- def tearDown(self):
- self.serv.close()
- self.serv = None
-
- def clientSetUp(self):
- pass
-
- def clientTearDown(self):
- self.cli.close()
- self.cli = None
- ThreadableTest.clientTearDown(self)
-
-
-#######################################################################
-## Begin Tests
-
-class GeneralModuleTests(unittest.TestCase):
-
- def test_weakref(self):
- if sys.platform[:4] == 'java': return
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- p = proxy(s)
- self.assertEqual(p.fileno(), s.fileno())
- s.close()
- s = None
- try:
- p.fileno()
- except ReferenceError:
- pass
- else:
- self.fail('Socket proxy still exists')
-
- def testSocketError(self):
- # Testing socket module exceptions
- def raise_error(*args, **kwargs):
- raise socket.error
- def raise_herror(*args, **kwargs):
- raise socket.herror
- def raise_gaierror(*args, **kwargs):
- raise socket.gaierror
- self.failUnlessRaises(socket.error, raise_error,
- "Error raising socket exception.")
- self.failUnlessRaises(socket.error, raise_herror,
- "Error raising socket exception.")
- self.failUnlessRaises(socket.error, raise_gaierror,
- "Error raising socket exception.")
-
- def testCrucialConstants(self):
- # Testing for mission critical constants
- socket.AF_INET
- socket.SOCK_STREAM
- socket.SOCK_DGRAM
- socket.SOCK_RAW
- socket.SOCK_RDM
- socket.SOCK_SEQPACKET
- socket.SOL_SOCKET
- socket.SO_REUSEADDR
-
- def testHostnameRes(self):
- # Testing hostname resolution mechanisms
- hostname = socket.gethostname()
- try:
- ip = socket.gethostbyname(hostname)
- except socket.error:
- # Probably name lookup wasn't set up right; skip this test
- self.fail("Probably name lookup wasn't set up right; skip testHostnameRes.gethostbyname")
- return
- self.assert_(ip.find('.') >= 0, "Error resolving host to ip.")
- try:
- hname, aliases, ipaddrs = socket.gethostbyaddr(ip)
- except socket.error:
- # Probably a similar problem as above; skip this test
- self.fail("Probably name lookup wasn't set up right; skip testHostnameRes.gethostbyaddr")
- return
- all_host_names = [hostname, hname] + aliases
- fqhn = socket.getfqdn()
- if not fqhn in all_host_names:
- self.fail("Error testing host resolution mechanisms.")
-
- def testRefCountGetNameInfo(self):
- # Testing reference count for getnameinfo
- import sys
- if hasattr(sys, "getrefcount"):
- try:
- # On some versions, this loses a reference
- orig = sys.getrefcount(__name__)
- socket.getnameinfo(__name__,0)
- except SystemError:
- if sys.getrefcount(__name__) <> orig:
- self.fail("socket.getnameinfo loses a reference")
-
- def testInterpreterCrash(self):
- if sys.platform[:4] == 'java': return
- # Making sure getnameinfo doesn't crash the interpreter
- try:
- # On some versions, this crashes the interpreter.
- socket.getnameinfo(('x', 0, 0, 0), 0)
- except socket.error:
- pass
-
-# Need to implement binary AND for ints and longs
-
- def testNtoH(self):
- if sys.platform[:4] == 'java': return # problems with int & long
- # This just checks that htons etc. are their own inverse,
- # when looking at the lower 16 or 32 bits.
- sizes = {socket.htonl: 32, socket.ntohl: 32,
- socket.htons: 16, socket.ntohs: 16}
- for func, size in sizes.items():
- mask = (1L<<size) - 1
- for i in (0, 1, 0xffff, ~0xffff, 2, 0x01234567, 0x76543210):
- self.assertEqual(i & mask, func(func(i&mask)) & mask)
-
- swapped = func(mask)
- self.assertEqual(swapped & mask, mask)
- self.assertRaises(OverflowError, func, 1L<<34)
-
- def testGetServBy(self):
- if sys.platform[:4] == 'java': return # not implemented on java
- eq = self.assertEqual
- # Find one service that exists, then check all the related interfaces.
- # I've ordered this by protocols that have both a tcp and udp
- # protocol, at least for modern Linuxes.
- if sys.platform in ('linux2', 'freebsd4', 'freebsd5', 'freebsd6',
- 'darwin'):
- # avoid the 'echo' service on this platform, as there is an
- # assumption breaking non-standard port/protocol entry
- services = ('daytime', 'qotd', 'domain')
- else:
- services = ('echo', 'daytime', 'domain')
- for service in services:
- try:
- port = socket.getservbyname(service, 'tcp')
- break
- except socket.error:
- pass
- else:
- raise socket.error
- # 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:
- udpport = None
- else:
- eq(udpport, port)
- # Now make sure the lookup by port returns the same service name
- eq(socket.getservbyport(port2), service)
- eq(socket.getservbyport(port, 'tcp'), service)
- if udpport is not None:
- eq(socket.getservbyport(udpport, 'udp'), service)
-
- def testDefaultTimeout(self):
- # Testing default timeout
- # The default timeout should initially be None
- self.assertEqual(socket.getdefaulttimeout(), None)
- s = socket.socket()
- self.assertEqual(s.gettimeout(), None)
- s.close()
-
- # Set the default timeout to 10, and see if it propagates
- socket.setdefaulttimeout(10)
- self.assertEqual(socket.getdefaulttimeout(), 10)
- s = socket.socket()
- self.assertEqual(s.gettimeout(), 10)
- s.close()
-
- # Reset the default timeout to None, and see if it propagates
- socket.setdefaulttimeout(None)
- self.assertEqual(socket.getdefaulttimeout(), None)
- s = socket.socket()
- self.assertEqual(s.gettimeout(), None)
- s.close()
-
- # Check that setting it to an invalid value raises ValueError
- self.assertRaises(ValueError, socket.setdefaulttimeout, -1)
-
- # Check that setting it to an invalid type raises TypeError
- self.assertRaises(TypeError, socket.setdefaulttimeout, "spam")
-
- def testIPv4toString(self):
- if not hasattr(socket, 'inet_pton'):
- return # No inet_pton() on this platform
- from socket import inet_aton as f, inet_pton, AF_INET
- g = lambda a: inet_pton(AF_INET, a)
-
- self.assertEquals('\x00\x00\x00\x00', f('0.0.0.0'))
- self.assertEquals('\xff\x00\xff\x00', f('255.0.255.0'))
- self.assertEquals('\xaa\xaa\xaa\xaa', f('170.170.170.170'))
- self.assertEquals('\x01\x02\x03\x04', f('1.2.3.4'))
-
- self.assertEquals('\x00\x00\x00\x00', g('0.0.0.0'))
- self.assertEquals('\xff\x00\xff\x00', g('255.0.255.0'))
- self.assertEquals('\xaa\xaa\xaa\xaa', g('170.170.170.170'))
-
- def testIPv6toString(self):
- if not hasattr(socket, 'inet_pton'):
- return # No inet_pton() on this platform
- try:
- from socket import inet_pton, AF_INET6, has_ipv6
- if not has_ipv6:
- return
- except ImportError:
- return
- f = lambda a: inet_pton(AF_INET6, a)
-
- self.assertEquals('\x00' * 16, f('::'))
- self.assertEquals('\x00' * 16, f('0::0'))
- self.assertEquals('\x00\x01' + '\x00' * 14, f('1::'))
- self.assertEquals(
- '\x45\xef\x76\xcb\x00\x1a\x56\xef\xaf\xeb\x0b\xac\x19\x24\xae\xae',
- f('45ef:76cb:1a:56ef:afeb:bac:1924:aeae')
- )
-
- def testStringToIPv4(self):
- if not hasattr(socket, 'inet_ntop'):
- return # No inet_ntop() on this platform
- from socket import inet_ntoa as f, inet_ntop, AF_INET
- g = lambda a: inet_ntop(AF_INET, a)
-
- self.assertEquals('1.0.1.0', f('\x01\x00\x01\x00'))
- self.assertEquals('170.85.170.85', f('\xaa\x55\xaa\x55'))
- self.assertEquals('255.255.255.255', f('\xff\xff\xff\xff'))
- self.assertEquals('1.2.3.4', f('\x01\x02\x03\x04'))
-
- self.assertEquals('1.0.1.0', g('\x01\x00\x01\x00'))
- self.assertEquals('170.85.170.85', g('\xaa\x55\xaa\x55'))
- self.assertEquals('255.255.255.255', g('\xff\xff\xff\xff'))
-
- def testStringToIPv6(self):
- if not hasattr(socket, 'inet_ntop'):
- return # No inet_ntop() on this platform
- try:
- from socket import inet_ntop, AF_INET6, has_ipv6
- if not has_ipv6:
- return
- except ImportError:
- return
- f = lambda a: inet_ntop(AF_INET6, a)
-
- self.assertEquals('::', f('\x00' * 16))
- self.assertEquals('::1', f('\x00' * 15 + '\x01'))
- self.assertEquals(
- 'aef:b01:506:1001:ffff:9997:55:170',
- f('\x0a\xef\x0b\x01\x05\x06\x10\x01\xff\xff\x99\x97\x00\x55\x01\x70')
- )
-
- # XXX The following don't test module-level functionality...
-
- def testSockName(self):
- # Testing getsockname()
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.bind(("0.0.0.0", PORT+1))
- name = sock.getsockname()
- self.assertEqual(name, ("0.0.0.0", PORT+1))
-
- def testGetSockOpt(self):
- # Testing getsockopt()
- # We know a socket should start without reuse==0
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
- self.failIf(reuse != 0, "initial mode is reuse")
-
- def testSetSockOpt(self):
- # Testing setsockopt()
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
- reuse = sock.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
- self.failIf(reuse == 0, "failed to set reuse mode")
-
- def testSendAfterClose(self):
- # testing send() after close() with timeout
- sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- sock.settimeout(1)
- sock.close()
- self.assertRaises(socket.error, sock.send, "spam")
-
-class BasicTCPTest(SocketConnectedTest):
-
- def __init__(self, methodName='runTest'):
- SocketConnectedTest.__init__(self, methodName=methodName)
-
- def testRecv(self):
- # Testing large receive over TCP
- msg = self.cli_conn.recv(1024)
- self.assertEqual(msg, MSG)
-
- def _testRecv(self):
- self.serv_conn.send(MSG)
-
- def testOverFlowRecv(self):
- # Testing receive in chunks over TCP
- seg1 = self.cli_conn.recv(len(MSG) - 3)
- seg2 = self.cli_conn.recv(1024)
- msg = seg1 + seg2
- self.assertEqual(msg, MSG)
-
- def _testOverFlowRecv(self):
- self.serv_conn.send(MSG)
-
- def testRecvFrom(self):
- # Testing large recvfrom() over TCP
- msg, addr = self.cli_conn.recvfrom(1024)
- self.assertEqual(msg, MSG)
-
- def _testRecvFrom(self):
- self.serv_conn.send(MSG)
-
- def testOverFlowRecvFrom(self):
- # Testing recvfrom() in chunks over TCP
- seg1, addr = self.cli_conn.recvfrom(len(MSG)-3)
- seg2, addr = self.cli_conn.recvfrom(1024)
- msg = seg1 + seg2
- self.assertEqual(msg, MSG)
-
- def _testOverFlowRecvFrom(self):
- self.serv_conn.send(MSG)
-
- def testSendAll(self):
- # Testing sendall() with a 2048 byte string over TCP
- msg = ''
- while 1:
- read = self.cli_conn.recv(1024)
- if not read:
- break
- msg += read
- self.assertEqual(msg, 'f' * 2048)
-
- def _testSendAll(self):
- big_chunk = 'f' * 2048
- self.serv_conn.sendall(big_chunk)
-
- def testFromFd(self):
- # Testing fromfd()
- if not hasattr(socket, "fromfd"):
- return # On Windows, this doesn't exist
- fd = self.cli_conn.fileno()
- sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
- msg = sock.recv(1024)
- self.assertEqual(msg, MSG)
-
- def _testFromFd(self):
- self.serv_conn.send(MSG)
-
- def testShutdown(self):
- # Testing shutdown()
- msg = self.cli_conn.recv(1024)
- self.assertEqual(msg, MSG)
-
- def _testShutdown(self):
- self.serv_conn.send(MSG)
- self.serv_conn.shutdown(2)
-
-class BasicUDPTest(ThreadedUDPSocketTest):
-
- def __init__(self, methodName='runTest'):
- ThreadedUDPSocketTest.__init__(self, methodName=methodName)
-
- def testSendtoAndRecv(self):
- # Testing sendto() and Recv() over UDP
- msg = self.serv.recv(len(MSG))
- self.assertEqual(msg, MSG)
-
- def _testSendtoAndRecv(self):
- self.cli.sendto(MSG, 0, (HOST, PORT))
-
- def testRecvFrom(self):
- # Testing recvfrom() over UDP
- msg, addr = self.serv.recvfrom(len(MSG))
- self.assertEqual(msg, MSG)
-
- def _testRecvFrom(self):
- self.cli.sendto(MSG, 0, (HOST, PORT))
-
- def testSendtoEightBitSafe(self):
- # This test is necessary because java only supports signed bytes
- msg = self.serv.recv(len(EIGHT_BIT_MSG))
- self.assertEqual(msg, EIGHT_BIT_MSG)
-
- def _testSendtoEightBitSafe(self):
- self.cli.sendto(EIGHT_BIT_MSG, 0, (HOST, PORT))
-
-class BasicSocketPairTest(SocketPairTest):
-
- def __init__(self, methodName='runTest'):
- SocketPairTest.__init__(self, methodName=methodName)
-
- def testRecv(self):
- msg = self.serv.recv(1024)
- self.assertEqual(msg, MSG)
-
- def _testRecv(self):
- self.cli.send(MSG)
-
- def testSend(self):
- self.serv.send(MSG)
-
- def _testSend(self):
- msg = self.cli.recv(1024)
- self.assertEqual(msg, MSG)
-
-class NonBlockingTCPTests(ThreadedTCPSocketTest):
-
- def __init__(self, methodName='runTest'):
- ThreadedTCPSocketTest.__init__(self, methodName=methodName)
-
- def testSetBlocking(self):
- # Testing whether set blocking works
- self.serv.setblocking(0)
- start = time.time()
- try:
- self.serv.accept()
- except socket.error:
- pass
- end = time.time()
- self.assert_((end - start) < 1.0, "Error setting non-blocking mode.")
-
- def _testSetBlocking(self):
- pass
-
- #
- # AMAK: 20070307
- # Split testAccept into two separate tests
- # 1. A test for non-blocking accept when there is NO connection pending
- # 2. A test for non-blocking accept when there is A connection pending
- # I think that perhaps the only reason the original combined test passes
- # on cpython is because of thread timing and sychronization parameters
- # of that platform.
- #
-
- def testAcceptNoConnection(self):
- # Testing non-blocking accept returns immediately when no connection
- self.serv.setblocking(0)
- try:
- conn, addr = self.serv.accept()
- except socket.error:
- pass
- else:
- self.fail("Error trying to do non-blocking accept.")
-
- def _testAcceptNoConnection(self):
- # Client side does nothing
- pass
-
- def testAcceptConnection(self):
- # Testing non-blocking accept works when connection present
- self.serv.setblocking(0)
- read, write, err = select.select([self.serv], [], [])
- if self.serv in read:
- conn, addr = self.serv.accept()
- else:
- self.fail("Error trying to do accept after select: server socket was not in 'read'able list")
-
- def _testAcceptConnection(self):
- # Make a connection to the server
- self.cli.connect((HOST, PORT))
-
- #
- # AMAK: 20070311
- # Introduced a new test for non-blocking connect
- # Renamed old testConnect to testBlockingConnect
- #
-
- def testBlockingConnect(self):
- # Testing blocking connect
- conn, addr = self.serv.accept()
-
- def _testBlockingConnect(self):
- # Testing blocking connect
- self.cli.settimeout(10)
- self.cli.connect((HOST, PORT))
-
- def testNonBlockingConnect(self):
- # Testing non-blocking connect
- conn, addr = self.serv.accept()
-
- def _testNonBlockingConnect(self):
- # Testing non-blocking connect
- self.cli.setblocking(0)
- result = self.cli.connect_ex((HOST, PORT))
- rfds, wfds, xfds = select.select([], [self.cli], [])
- self.failUnless(self.cli in wfds)
- try:
- self.cli.send(MSG)
- except socket.error:
- self.fail("Sending on connected socket should not have raised socket.error")
-
- #
- # AMAK: 20070518
- # Introduced a new test for connect with bind to specific local address
- #
-
- def testConnectWithLocalBind(self):
- # Test blocking connect
- conn, addr = self.serv.accept()
-
- def _testConnectWithLocalBind(self):
- # Testing blocking connect with local bind
- self.cli.settimeout(1)
- self.cli.bind( (HOST, PORT-1) )
- self.cli.connect((HOST, PORT))
- bound_host, bound_port = self.cli.getsockname()
- self.failUnlessEqual(bound_port, PORT-1)
-
- def testRecvData(self):
- # Testing non-blocking recv
- conn, addr = self.serv.accept()
- conn.setblocking(0)
- rfds, wfds, xfds = select.select([conn], [], [])
- if conn in rfds:
- msg = conn.recv(len(MSG))
- self.assertEqual(msg, MSG)
- else:
- self.fail("Non-blocking socket with data should been in read list.")
-
- def _testRecvData(self):
- self.cli.connect((HOST, PORT))
- self.cli.send(MSG)
-
- def testRecvNoData(self):
- # Testing non-blocking recv
- conn, addr = self.serv.accept()
- conn.setblocking(0)
- try:
- msg = conn.recv(len(MSG))
- except socket.error:
- pass
- else:
- self.fail("Non-blocking recv of no data should have raised socket.error.")
-
- def _testRecvNoData(self):
- self.cli.connect((HOST, PORT))
- time.sleep(0.1)
-
-class NonBlockingUDPTests(ThreadedUDPSocketTest): pass
-
-#
-# TODO: Write some non-blocking UDP tests
-#
-
-class FileObjectClassTestCase(SocketConnectedTest):
-
- bufsize = -1 # Use default buffer size
-
- def __init__(self, methodName='runTest'):
- SocketConnectedTest.__init__(self, methodName=methodName)
-
- def setUp(self):
- SocketConnectedTest.setUp(self)
- self.serv_file = self.cli_conn.makefile('rb', self.bufsize)
-
- def tearDown(self):
- self.serv_file.close()
- self.assert_(self.serv_file.closed)
- self.serv_file = None
- SocketConnectedTest.tearDown(self)
-
- def clientSetUp(self):
- SocketConnectedTest.clientSetUp(self)
- self.cli_file = self.serv_conn.makefile('wb')
-
- def clientTearDown(self):
- self.cli_file.close()
- self.assert_(self.cli_file.closed)
- self.cli_file = None
- SocketConnectedTest.clientTearDown(self)
-
- def testSmallRead(self):
- # Performing small file read test
- first_seg = self.serv_file.read(len(MSG)-3)
- second_seg = self.serv_file.read(3)
- msg = first_seg + second_seg
- self.assertEqual(msg, MSG)
-
- def _testSmallRead(self):
- self.cli_file.write(MSG)
- self.cli_file.flush()
-
- def testFullRead(self):
- # read until EOF
- msg = self.serv_file.read()
- self.assertEqual(msg, MSG)
-
- def _testFullRead(self):
- self.cli_file.write(MSG)
- self.cli_file.close()
-
- def testUnbufferedRead(self):
- # Performing unbuffered file read test
- buf = ''
- while 1:
- char = self.serv_file.read(1)
- if not char:
- break
- buf += char
- self.assertEqual(buf, MSG)
-
- def _testUnbufferedRead(self):
- self.cli_file.write(MSG)
- self.cli_file.flush()
-
- def testReadline(self):
- # Performing file readline test
- line = self.serv_file.readline()
- self.assertEqual(line, MSG)
-
- def _testReadline(self):
- self.cli_file.write(MSG)
- self.cli_file.flush()
-
- def testClosedAttr(self):
- self.assert_(not self.serv_file.closed)
-
- def _testClosedAttr(self):
- self.assert_(not self.cli_file.closed)
-
-class UnbufferedFileObjectClassTestCase(FileObjectClassTestCase):
-
- """Repeat the tests from FileObjectClassTestCase with bufsize==0.
-
- In this case (and in this case only), it should be possible to
- create a file object, read a line from it, create another file
- object, read another line from it, without loss of data in the
- first file object's buffer. Note that httplib relies on this
- when reading multiple requests from the same socket."""
-
- bufsize = 0 # Use unbuffered mode
-
- def testUnbufferedReadline(self):
- # Read a line, create a new file object, read another line with it
- line = self.serv_file.readline() # first line
- self.assertEqual(line, "A. " + MSG) # first line
- self.serv_file = self.cli_conn.makefile('rb', 0)
- line = self.serv_file.readline() # second line
- self.assertEqual(line, "B. " + MSG) # second line
-
- def _testUnbufferedReadline(self):
- self.cli_file.write("A. " + MSG)
- self.cli_file.write("B. " + MSG)
- self.cli_file.flush()
-
-class LineBufferedFileObjectClassTestCase(FileObjectClassTestCase):
-
- bufsize = 1 # Default-buffered for reading; line-buffered for writing
-
-
-class SmallBufferedFileObjectClassTestCase(FileObjectClassTestCase):
-
- bufsize = 2 # Exercise the buffering code
-
-class TCPTimeoutTest(SocketTCPTest):
-
- def testTCPTimeout(self):
- def raise_timeout(*args, **kwargs):
- self.serv.settimeout(1.0)
- self.serv.accept()
- self.failUnlessRaises(socket.timeout, raise_timeout,
- "Error generating a timeout exception (TCP)")
-
- def testTimeoutZero(self):
- ok = False
- try:
- self.serv.settimeout(0.0)
- foo = self.serv.accept()
- except socket.timeout:
- self.fail("caught timeout instead of error (TCP)")
- except socket.error:
- ok = True
- except Exception, x:
- self.fail("caught unexpected exception (TCP): %s" % str(x))
- if not ok:
- self.fail("accept() returned success when we did not expect it")
-
-class TCPClientTimeoutTest(ThreadedTCPSocketTest):
-
- def testTCPClientTimeout(self):
- pass # i.e. do not accept
-
- def _testTCPClientTimeout(self):
- try:
- self.cli.settimeout(0.1)
- self.cli.connect( (HOST, PORT) )
- except socket.timeout, st:
- pass
- except Exception, x:
- self.fail("Client socket timeout should have raised socket.timeout, not %s" % str(x))
- else:
- self.fail("Client socket timeout should have raised socket.timeout")
-
-#
-# AMAK: 20070307
-# Corrected the superclass of UDPTimeoutTest
-#
-
-class UDPTimeoutTest(SocketUDPTest):
-
- def testUDPTimeout(self):
- def raise_timeout(*args, **kwargs):
- self.serv.settimeout(1.0)
- self.serv.recv(1024)
- self.failUnlessRaises(socket.timeout, raise_timeout,
- "Error generating a timeout exception (UDP)")
-
- def testTimeoutZero(self):
- ok = False
- try:
- self.serv.settimeout(0.0)
- foo = self.serv.recv(1024)
- except socket.timeout:
- self.fail("caught timeout instead of error (UDP)")
- except socket.error:
- ok = True
- except Exception, x:
- self.fail("caught unexpected exception (UDP): %s" % str(x))
- if not ok:
- self.fail("recv() returned success when we did not expect it")
-
-class TestExceptions(unittest.TestCase):
-
- def testExceptionTree(self):
- self.assert_(issubclass(socket.error, Exception))
- self.assert_(issubclass(socket.herror, socket.error))
- self.assert_(issubclass(socket.gaierror, socket.error))
- self.assert_(issubclass(socket.timeout, socket.error))
-
-class TestJythonExceptions(unittest.TestCase):
-
- def testHostNotFound(self):
- try:
- socket.gethostbyname("doesnotexist")
- except socket.gaierror, gaix:
- self.failUnlessEqual(gaix[0], errno.EGETADDRINFOFAILED)
- except Exception, x:
- self.fail("Get host name for non-existent host raised wrong exception: %s" % x)
-
- def testConnectionRefused(self):
- try:
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- # This port should not be open at this time
- s.connect( (HOST, PORT) )
- except socket.error, se:
- self.failUnlessEqual(se[0], errno.ECONNREFUSED)
- except Exception, x:
- self.fail("Connection to non-existent host/port raised wrong exception: %s" % x)
- else:
- self.fail("Socket (%s,%s) should not have been listening at this time" % (HOST, PORT))
-
- def testBindException(self):
- # First bind to the target port
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.bind( (HOST, PORT) )
- s.listen()
- try:
- try:
- # And then try to bind again
- t = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- t.bind( (HOST, PORT) )
- t.listen()
- except socket.error, se:
- self.failUnlessEqual(se[0], errno.EACCES)
- except Exception, x:
- self.fail("Binding to already bound host/port raised wrong exception: %s" % x)
- else:
- self.fail("Binding to already bound host/port should have raised exception")
- finally:
- s.close()
-
- def testUnresolvedAddress(self):
- try:
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.connect( ('non.existent.server', PORT) )
- except socket.gaierror, gaix:
- self.failUnlessEqual(gaix[0], errno.EGETADDRINFOFAILED)
- except Exception, x:
- self.fail("Get host name for non-existent host raised wrong exception: %s" % x)
- else:
- self.fail("Get host name for non-existent host should have raised exception")
-
-class TestAddressParameters:
-
- def testBindNonTupleEndpointRaisesTypeError(self):
- try:
- self.socket.bind(HOST, PORT)
- except TypeError:
- pass
- else:
- self.fail("Illegal non-tuple bind address did not raise TypeError")
-
- def testConnectNonTupleEndpointRaisesTypeError(self):
- try:
- self.socket.connect(HOST, PORT)
- except TypeError:
- pass
- else:
- self.fail("Illegal non-tuple connect address did not raise TypeError")
-
- def testConnectExNonTupleEndpointRaisesTypeError(self):
- try:
- self.socket.connect_ex(HOST, PORT)
- except TypeError:
- pass
- else:
- self.fail("Illegal non-tuple connect address did not raise TypeError")
-
-class TestTCPAddressParameters(unittest.TestCase, TestAddressParameters):
-
- def setUp(self):
- self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-
-class TestUDPAddressParameters(unittest.TestCase, TestAddressParameters):
-
- def setUp(self):
- self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
-
-def test_main():
- tests = [
- GeneralModuleTests,
- BasicTCPTest,
- TCPTimeoutTest,
- TCPClientTimeoutTest,
- TestExceptions,
- TestTCPAddressParameters,
- TestUDPAddressParameters,
- BasicUDPTest,
- UDPTimeoutTest,
- NonBlockingTCPTests,
- NonBlockingUDPTests,
- FileObjectClassTestCase,
- UnbufferedFileObjectClassTestCase,
- LineBufferedFileObjectClassTestCase,
- SmallBufferedFileObjectCla...
[truncated message content] |