[Jython-checkins] jython: Re-adding latest cpython 2.7 asyncore.py and test_asyncore.py from
alan.kennedy
jython-checkins at python.org
Sat Aug 25 22:49:23 CEST 2012
http://hg.python.org/jython/rev/7ea49e16dd54
changeset: 6853:7ea49e16dd54
user: Alan Kennedy <alan at xhaus.com>
date: Sat Aug 25 18:46:29 2012 +0100
summary:
Re-adding latest cpython 2.7 asyncore.py and test_asyncore.py from http://hg.python.org/cpython/file/c1c45755397b
files:
Lib/asyncore.py | 659 +++++++++++++++++++++++
Lib/test/test_asyncore.py | 743 ++++++++++++++++++++++++++
2 files changed, 1402 insertions(+), 0 deletions(-)
diff --git a/Lib/asyncore.py b/Lib/asyncore.py
new file mode 100644
--- /dev/null
+++ b/Lib/asyncore.py
@@ -0,0 +1,659 @@
+# -*- Mode: Python -*-
+# Id: asyncore.py,v 2.51 2000/09/07 22:29:26 rushing Exp
+# Author: Sam Rushing <rushing at nightmare.com>
+
+# ======================================================================
+# Copyright 1996 by Sam Rushing
+#
+# All Rights Reserved
+#
+# Permission to use, copy, modify, and distribute this software and
+# its documentation for any purpose and without fee is hereby
+# granted, provided that the above copyright notice appear in all
+# copies and that both that copyright notice and this permission
+# notice appear in supporting documentation, and that the name of Sam
+# Rushing not be used in advertising or publicity pertaining to
+# distribution of the software without specific, written prior
+# permission.
+#
+# SAM RUSHING DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE,
+# INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS, IN
+# NO EVENT SHALL SAM RUSHING BE LIABLE FOR ANY SPECIAL, INDIRECT OR
+# CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS
+# OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
+# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
+# CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
+# ======================================================================
+
+"""Basic infrastructure for asynchronous socket service clients and servers.
+
+There are only two ways to have a program on a single processor do "more
+than one thing at a time". Multi-threaded programming is the simplest and
+most popular way to do it, but there is another very different technique,
+that lets you have nearly all the advantages of multi-threading, without
+actually using multiple threads. it's really only practical if your program
+is largely I/O bound. If your program is CPU bound, then pre-emptive
+scheduled threads are probably what you really need. Network servers are
+rarely CPU-bound, however.
+
+If your operating system supports the select() system call in its I/O
+library (and nearly all do), then you can use it to juggle multiple
+communication channels at once; doing other work while your I/O is taking
+place in the "background." Although this strategy can seem strange and
+complex, especially at first, it is in many ways easier to understand and
+control than multi-threaded programming. The module documented here solves
+many of the difficult problems for you, making the task of building
+sophisticated high-performance network servers and clients a snap.
+"""
+
+import select
+import socket
+import sys
+import time
+import warnings
+
+import os
+from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, EINVAL, \
+ ENOTCONN, ESHUTDOWN, EINTR, EISCONN, EBADF, ECONNABORTED, EPIPE, EAGAIN, \
+ errorcode
+
+_DISCONNECTED = frozenset((ECONNRESET, ENOTCONN, ESHUTDOWN, ECONNABORTED, EPIPE,
+ EBADF))
+
+try:
+ socket_map
+except NameError:
+ socket_map = {}
+
+def _strerror(err):
+ try:
+ return os.strerror(err)
+ except (ValueError, OverflowError, NameError):
+ if err in errorcode:
+ return errorcode[err]
+ return "Unknown error %s" %err
+
+class ExitNow(Exception):
+ pass
+
+_reraised_exceptions = (ExitNow, KeyboardInterrupt, SystemExit)
+
+def read(obj):
+ try:
+ obj.handle_read_event()
+ except _reraised_exceptions:
+ raise
+ except:
+ obj.handle_error()
+
+def write(obj):
+ try:
+ obj.handle_write_event()
+ except _reraised_exceptions:
+ raise
+ except:
+ obj.handle_error()
+
+def _exception(obj):
+ try:
+ obj.handle_expt_event()
+ except _reraised_exceptions:
+ raise
+ except:
+ obj.handle_error()
+
+def readwrite(obj, flags):
+ try:
+ if flags & select.POLLIN:
+ obj.handle_read_event()
+ if flags & select.POLLOUT:
+ obj.handle_write_event()
+ if flags & select.POLLPRI:
+ obj.handle_expt_event()
+ if flags & (select.POLLHUP | select.POLLERR | select.POLLNVAL):
+ obj.handle_close()
+ except socket.error, e:
+ if e.args[0] not in _DISCONNECTED:
+ obj.handle_error()
+ else:
+ obj.handle_close()
+ except _reraised_exceptions:
+ raise
+ except:
+ obj.handle_error()
+
+def poll(timeout=0.0, map=None):
+ if map is None:
+ map = socket_map
+ if map:
+ r = []; w = []; e = []
+ for fd, obj in map.items():
+ is_r = obj.readable()
+ is_w = obj.writable()
+ if is_r:
+ r.append(fd)
+ # accepting sockets should not be writable
+ if is_w and not obj.accepting:
+ w.append(fd)
+ if is_r or is_w:
+ e.append(fd)
+ if [] == r == w == e:
+ time.sleep(timeout)
+ return
+
+ try:
+ r, w, e = select.select(r, w, e, timeout)
+ except select.error, err:
+ if err.args[0] != EINTR:
+ raise
+ else:
+ return
+
+ for fd in r:
+ obj = map.get(fd)
+ if obj is None:
+ continue
+ read(obj)
+
+ for fd in w:
+ obj = map.get(fd)
+ if obj is None:
+ continue
+ write(obj)
+
+ for fd in e:
+ obj = map.get(fd)
+ if obj is None:
+ continue
+ _exception(obj)
+
+def poll2(timeout=0.0, map=None):
+ # Use the poll() support added to the select module in Python 2.0
+ if map is None:
+ map = socket_map
+ if timeout is not None:
+ # timeout is in milliseconds
+ timeout = int(timeout*1000)
+ pollster = select.poll()
+ if map:
+ for fd, obj in map.items():
+ flags = 0
+ if obj.readable():
+ flags |= select.POLLIN | select.POLLPRI
+ # accepting sockets should not be writable
+ if obj.writable() and not obj.accepting:
+ flags |= select.POLLOUT
+ if flags:
+ # Only check for exceptions if object was either readable
+ # or writable.
+ flags |= select.POLLERR | select.POLLHUP | select.POLLNVAL
+ pollster.register(fd, flags)
+ try:
+ r = pollster.poll(timeout)
+ except select.error, err:
+ if err.args[0] != EINTR:
+ raise
+ r = []
+ for fd, flags in r:
+ obj = map.get(fd)
+ if obj is None:
+ continue
+ readwrite(obj, flags)
+
+poll3 = poll2 # Alias for backward compatibility
+
+def loop(timeout=30.0, use_poll=False, map=None, count=None):
+ if map is None:
+ map = socket_map
+
+ if use_poll and hasattr(select, 'poll'):
+ poll_fun = poll2
+ else:
+ poll_fun = poll
+
+ if count is None:
+ while map:
+ poll_fun(timeout, map)
+
+ else:
+ while map and count > 0:
+ poll_fun(timeout, map)
+ count = count - 1
+
+class dispatcher:
+
+ debug = False
+ connected = False
+ accepting = False
+ connecting = False
+ closing = False
+ addr = None
+ ignore_log_types = frozenset(['warning'])
+
+ def __init__(self, sock=None, map=None):
+ if map is None:
+ self._map = socket_map
+ else:
+ self._map = map
+
+ self._fileno = None
+
+ if sock:
+ # Set to nonblocking just to make sure for cases where we
+ # get a socket from a blocking source.
+ sock.setblocking(0)
+ self.set_socket(sock, map)
+ self.connected = True
+ # The constructor no longer requires that the socket
+ # passed be connected.
+ try:
+ self.addr = sock.getpeername()
+ except socket.error, err:
+ if err.args[0] in (ENOTCONN, EINVAL):
+ # To handle the case where we got an unconnected
+ # socket.
+ self.connected = False
+ else:
+ # The socket is broken in some unknown way, alert
+ # the user and remove it from the map (to prevent
+ # polling of broken sockets).
+ self.del_channel(map)
+ raise
+ else:
+ self.socket = None
+
+ def __repr__(self):
+ status = [self.__class__.__module__+"."+self.__class__.__name__]
+ if self.accepting and self.addr:
+ status.append('listening')
+ elif self.connected:
+ status.append('connected')
+ if self.addr is not None:
+ try:
+ status.append('%s:%d' % self.addr)
+ except TypeError:
+ status.append(repr(self.addr))
+ return '<%s at %#x>' % (' '.join(status), id(self))
+
+ __str__ = __repr__
+
+ def add_channel(self, map=None):
+ #self.log_info('adding channel %s' % self)
+ if map is None:
+ map = self._map
+ map[self._fileno] = self
+
+ def del_channel(self, map=None):
+ fd = self._fileno
+ if map is None:
+ map = self._map
+ if fd in map:
+ #self.log_info('closing channel %d:%s' % (fd, self))
+ del map[fd]
+ self._fileno = None
+
+ def create_socket(self, family, type):
+ self.family_and_type = family, type
+ sock = socket.socket(family, type)
+ sock.setblocking(0)
+ self.set_socket(sock)
+
+ def set_socket(self, sock, map=None):
+ self.socket = sock
+## self.__dict__['socket'] = sock
+ self._fileno = sock.fileno()
+ self.add_channel(map)
+
+ def set_reuse_addr(self):
+ # try to re-use a server port if possible
+ try:
+ self.socket.setsockopt(
+ socket.SOL_SOCKET, socket.SO_REUSEADDR,
+ self.socket.getsockopt(socket.SOL_SOCKET,
+ socket.SO_REUSEADDR) | 1
+ )
+ except socket.error:
+ pass
+
+ # ==================================================
+ # predicates for select()
+ # these are used as filters for the lists of sockets
+ # to pass to select().
+ # ==================================================
+
+ def readable(self):
+ return True
+
+ def writable(self):
+ return True
+
+ # ==================================================
+ # socket object methods.
+ # ==================================================
+
+ def listen(self, num):
+ self.accepting = True
+ if os.name == 'nt' and num > 5:
+ num = 5
+ return self.socket.listen(num)
+
+ def bind(self, addr):
+ self.addr = addr
+ return self.socket.bind(addr)
+
+ def connect(self, address):
+ self.connected = False
+ self.connecting = True
+ err = self.socket.connect_ex(address)
+ if err in (EINPROGRESS, EALREADY, EWOULDBLOCK) \
+ or err == EINVAL and os.name in ('nt', 'ce'):
+ self.addr = address
+ return
+ if err in (0, EISCONN):
+ self.addr = address
+ self.handle_connect_event()
+ else:
+ raise socket.error(err, errorcode[err])
+
+ def accept(self):
+ # XXX can return either an address pair or None
+ try:
+ conn, addr = self.socket.accept()
+ except TypeError:
+ return None
+ except socket.error as why:
+ if why.args[0] in (EWOULDBLOCK, ECONNABORTED, EAGAIN):
+ return None
+ else:
+ raise
+ else:
+ return conn, addr
+
+ def send(self, data):
+ try:
+ result = self.socket.send(data)
+ return result
+ except socket.error, why:
+ if why.args[0] == EWOULDBLOCK:
+ return 0
+ elif why.args[0] in _DISCONNECTED:
+ self.handle_close()
+ return 0
+ else:
+ raise
+
+ def recv(self, buffer_size):
+ try:
+ data = self.socket.recv(buffer_size)
+ if not data:
+ # a closed connection is indicated by signaling
+ # a read condition, and having recv() return 0.
+ self.handle_close()
+ return ''
+ else:
+ return data
+ except socket.error, why:
+ # winsock sometimes throws ENOTCONN
+ if why.args[0] in _DISCONNECTED:
+ self.handle_close()
+ return ''
+ else:
+ raise
+
+ def close(self):
+ self.connected = False
+ self.accepting = False
+ self.connecting = False
+ self.del_channel()
+ try:
+ self.socket.close()
+ except socket.error, why:
+ if why.args[0] not in (ENOTCONN, EBADF):
+ raise
+
+ # cheap inheritance, used to pass all other attribute
+ # references to the underlying socket object.
+ def __getattr__(self, attr):
+ try:
+ retattr = getattr(self.socket, attr)
+ except AttributeError:
+ raise AttributeError("%s instance has no attribute '%s'"
+ %(self.__class__.__name__, attr))
+ else:
+ msg = "%(me)s.%(attr)s is deprecated. Use %(me)s.socket.%(attr)s " \
+ "instead." % {'me': self.__class__.__name__, 'attr':attr}
+ warnings.warn(msg, DeprecationWarning, stacklevel=2)
+ return retattr
+
+ # log and log_info may be overridden to provide more sophisticated
+ # logging and warning methods. In general, log is for 'hit' logging
+ # and 'log_info' is for informational, warning and error logging.
+
+ def log(self, message):
+ sys.stderr.write('log: %s\n' % str(message))
+
+ def log_info(self, message, type='info'):
+ if type not in self.ignore_log_types:
+ print '%s: %s' % (type, message)
+
+ def handle_read_event(self):
+ if self.accepting:
+ # accepting sockets are never connected, they "spawn" new
+ # sockets that are connected
+ self.handle_accept()
+ elif not self.connected:
+ if self.connecting:
+ self.handle_connect_event()
+ self.handle_read()
+ else:
+ self.handle_read()
+
+ def handle_connect_event(self):
+ err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
+ if err != 0:
+ raise socket.error(err, _strerror(err))
+ self.handle_connect()
+ self.connected = True
+ self.connecting = False
+
+ def handle_write_event(self):
+ if self.accepting:
+ # Accepting sockets shouldn't get a write event.
+ # We will pretend it didn't happen.
+ return
+
+ if not self.connected:
+ if self.connecting:
+ self.handle_connect_event()
+ self.handle_write()
+
+ def handle_expt_event(self):
+ # handle_expt_event() is called if there might be an error on the
+ # socket, or if there is OOB data
+ # check for the error condition first
+ err = self.socket.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
+ if err != 0:
+ # we can get here when select.select() says that there is an
+ # exceptional condition on the socket
+ # since there is an error, we'll go ahead and close the socket
+ # like we would in a subclassed handle_read() that received no
+ # data
+ self.handle_close()
+ else:
+ self.handle_expt()
+
+ def handle_error(self):
+ nil, t, v, tbinfo = compact_traceback()
+
+ # sometimes a user repr method will crash.
+ try:
+ self_repr = repr(self)
+ except:
+ self_repr = '<__repr__(self) failed for object at %0x>' % id(self)
+
+ self.log_info(
+ 'uncaptured python exception, closing channel %s (%s:%s %s)' % (
+ self_repr,
+ t,
+ v,
+ tbinfo
+ ),
+ 'error'
+ )
+ self.handle_close()
+
+ def handle_expt(self):
+ self.log_info('unhandled incoming priority event', 'warning')
+
+ def handle_read(self):
+ self.log_info('unhandled read event', 'warning')
+
+ def handle_write(self):
+ self.log_info('unhandled write event', 'warning')
+
+ def handle_connect(self):
+ self.log_info('unhandled connect event', 'warning')
+
+ def handle_accept(self):
+ self.log_info('unhandled accept event', 'warning')
+
+ def handle_close(self):
+ self.log_info('unhandled close event', 'warning')
+ self.close()
+
+# ---------------------------------------------------------------------------
+# adds simple buffered output capability, useful for simple clients.
+# [for more sophisticated usage use asynchat.async_chat]
+# ---------------------------------------------------------------------------
+
+class dispatcher_with_send(dispatcher):
+
+ def __init__(self, sock=None, map=None):
+ dispatcher.__init__(self, sock, map)
+ self.out_buffer = ''
+
+ def initiate_send(self):
+ num_sent = 0
+ num_sent = dispatcher.send(self, self.out_buffer[:512])
+ self.out_buffer = self.out_buffer[num_sent:]
+
+ def handle_write(self):
+ self.initiate_send()
+
+ def writable(self):
+ return (not self.connected) or len(self.out_buffer)
+
+ def send(self, data):
+ if self.debug:
+ self.log_info('sending %s' % repr(data))
+ self.out_buffer = self.out_buffer + data
+ self.initiate_send()
+
+# ---------------------------------------------------------------------------
+# used for debugging.
+# ---------------------------------------------------------------------------
+
+def compact_traceback():
+ t, v, tb = sys.exc_info()
+ tbinfo = []
+ if not tb: # Must have a traceback
+ raise AssertionError("traceback does not exist")
+ while tb:
+ tbinfo.append((
+ tb.tb_frame.f_code.co_filename,
+ tb.tb_frame.f_code.co_name,
+ str(tb.tb_lineno)
+ ))
+ tb = tb.tb_next
+
+ # just to be safe
+ del tb
+
+ file, function, line = tbinfo[-1]
+ info = ' '.join(['[%s|%s|%s]' % x for x in tbinfo])
+ return (file, function, line), t, v, info
+
+def close_all(map=None, ignore_all=False):
+ if map is None:
+ map = socket_map
+ for x in map.values():
+ try:
+ x.close()
+ except OSError, x:
+ if x.args[0] == EBADF:
+ pass
+ elif not ignore_all:
+ raise
+ except _reraised_exceptions:
+ raise
+ except:
+ if not ignore_all:
+ raise
+ map.clear()
+
+# Asynchronous File I/O:
+#
+# After a little research (reading man pages on various unixen, and
+# digging through the linux kernel), I've determined that select()
+# isn't meant for doing asynchronous file i/o.
+# Heartening, though - reading linux/mm/filemap.c shows that linux
+# supports asynchronous read-ahead. So _MOST_ of the time, the data
+# will be sitting in memory for us already when we go to read it.
+#
+# What other OS's (besides NT) support async file i/o? [VMS?]
+#
+# Regardless, this is useful for pipes, and stdin/stdout...
+
+if os.name == 'posix':
+ import fcntl
+
+ class file_wrapper:
+ # Here we override just enough to make a file
+ # look like a socket for the purposes of asyncore.
+ # The passed fd is automatically os.dup()'d
+
+ def __init__(self, fd):
+ self.fd = os.dup(fd)
+
+ def recv(self, *args):
+ return os.read(self.fd, *args)
+
+ def send(self, *args):
+ return os.write(self.fd, *args)
+
+ def getsockopt(self, level, optname, buflen=None):
+ if (level == socket.SOL_SOCKET and
+ optname == socket.SO_ERROR and
+ not buflen):
+ return 0
+ raise NotImplementedError("Only asyncore specific behaviour "
+ "implemented.")
+
+ read = recv
+ write = send
+
+ def close(self):
+ os.close(self.fd)
+
+ def fileno(self):
+ return self.fd
+
+ class file_dispatcher(dispatcher):
+
+ def __init__(self, fd, map=None):
+ dispatcher.__init__(self, None, map)
+ self.connected = True
+ try:
+ fd = fd.fileno()
+ except AttributeError:
+ pass
+ self.set_file(fd)
+ # set it to non-blocking mode
+ flags = fcntl.fcntl(fd, fcntl.F_GETFL, 0)
+ flags = flags | os.O_NONBLOCK
+ fcntl.fcntl(fd, fcntl.F_SETFL, flags)
+
+ def set_file(self, fd):
+ self.socket = file_wrapper(fd)
+ self._fileno = self.socket.fileno()
+ self.add_channel()
diff --git a/Lib/test/test_asyncore.py b/Lib/test/test_asyncore.py
new file mode 100644
--- /dev/null
+++ b/Lib/test/test_asyncore.py
@@ -0,0 +1,743 @@
+import asyncore
+import unittest
+import select
+import os
+import socket
+import sys
+import time
+import warnings
+import errno
+import struct
+
+from test import test_support
+from test.test_support import TESTFN, run_unittest, unlink
+from StringIO import StringIO
+
+try:
+ import threading
+except ImportError:
+ threading = None
+
+HOST = test_support.HOST
+
+class dummysocket:
+ def __init__(self):
+ self.closed = False
+
+ def close(self):
+ self.closed = True
+
+ def fileno(self):
+ return 42
+
+class dummychannel:
+ def __init__(self):
+ self.socket = dummysocket()
+
+ def close(self):
+ self.socket.close()
+
+class exitingdummy:
+ def __init__(self):
+ pass
+
+ def handle_read_event(self):
+ raise asyncore.ExitNow()
+
+ handle_write_event = handle_read_event
+ handle_close = handle_read_event
+ handle_expt_event = handle_read_event
+
+class crashingdummy:
+ def __init__(self):
+ self.error_handled = False
+
+ def handle_read_event(self):
+ raise Exception()
+
+ handle_write_event = handle_read_event
+ handle_close = handle_read_event
+ handle_expt_event = handle_read_event
+
+ def handle_error(self):
+ self.error_handled = True
+
+# used when testing senders; just collects what it gets until newline is sent
+def capture_server(evt, buf, serv):
+ try:
+ serv.listen(5)
+ conn, addr = serv.accept()
+ except socket.timeout:
+ pass
+ else:
+ n = 200
+ while n > 0:
+ r, w, e = select.select([conn], [], [])
+ if r:
+ data = conn.recv(10)
+ # keep everything except for the newline terminator
+ buf.write(data.replace('\n', ''))
+ if '\n' in data:
+ break
+ n -= 1
+ time.sleep(0.01)
+
+ conn.close()
+ finally:
+ serv.close()
+ evt.set()
+
+
+class HelperFunctionTests(unittest.TestCase):
+ def test_readwriteexc(self):
+ # Check exception handling behavior of read, write and _exception
+
+ # check that ExitNow exceptions in the object handler method
+ # bubbles all the way up through asyncore read/write/_exception calls
+ tr1 = exitingdummy()
+ self.assertRaises(asyncore.ExitNow, asyncore.read, tr1)
+ self.assertRaises(asyncore.ExitNow, asyncore.write, tr1)
+ self.assertRaises(asyncore.ExitNow, asyncore._exception, tr1)
+
+ # check that an exception other than ExitNow in the object handler
+ # method causes the handle_error method to get called
+ tr2 = crashingdummy()
+ asyncore.read(tr2)
+ self.assertEqual(tr2.error_handled, True)
+
+ tr2 = crashingdummy()
+ asyncore.write(tr2)
+ self.assertEqual(tr2.error_handled, True)
+
+ tr2 = crashingdummy()
+ asyncore._exception(tr2)
+ self.assertEqual(tr2.error_handled, True)
+
+ # asyncore.readwrite uses constants in the select module that
+ # are not present in Windows systems (see this thread:
+ # http://mail.python.org/pipermail/python-list/2001-October/109973.html)
+ # These constants should be present as long as poll is available
+
+ @unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
+ def test_readwrite(self):
+ # Check that correct methods are called by readwrite()
+
+ attributes = ('read', 'expt', 'write', 'closed', 'error_handled')
+
+ expected = (
+ (select.POLLIN, 'read'),
+ (select.POLLPRI, 'expt'),
+ (select.POLLOUT, 'write'),
+ (select.POLLERR, 'closed'),
+ (select.POLLHUP, 'closed'),
+ (select.POLLNVAL, 'closed'),
+ )
+
+ class testobj:
+ def __init__(self):
+ self.read = False
+ self.write = False
+ self.closed = False
+ self.expt = False
+ self.error_handled = False
+
+ def handle_read_event(self):
+ self.read = True
+
+ def handle_write_event(self):
+ self.write = True
+
+ def handle_close(self):
+ self.closed = True
+
+ def handle_expt_event(self):
+ self.expt = True
+
+ def handle_error(self):
+ self.error_handled = True
+
+ for flag, expectedattr in expected:
+ tobj = testobj()
+ self.assertEqual(getattr(tobj, expectedattr), False)
+ asyncore.readwrite(tobj, flag)
+
+ # Only the attribute modified by the routine we expect to be
+ # called should be True.
+ for attr in attributes:
+ self.assertEqual(getattr(tobj, attr), attr==expectedattr)
+
+ # check that ExitNow exceptions in the object handler method
+ # bubbles all the way up through asyncore readwrite call
+ tr1 = exitingdummy()
+ self.assertRaises(asyncore.ExitNow, asyncore.readwrite, tr1, flag)
+
+ # check that an exception other than ExitNow in the object handler
+ # method causes the handle_error method to get called
+ tr2 = crashingdummy()
+ self.assertEqual(tr2.error_handled, False)
+ asyncore.readwrite(tr2, flag)
+ self.assertEqual(tr2.error_handled, True)
+
+ def test_closeall(self):
+ self.closeall_check(False)
+
+ def test_closeall_default(self):
+ self.closeall_check(True)
+
+ def closeall_check(self, usedefault):
+ # Check that close_all() closes everything in a given map
+
+ l = []
+ testmap = {}
+ for i in range(10):
+ c = dummychannel()
+ l.append(c)
+ self.assertEqual(c.socket.closed, False)
+ testmap[i] = c
+
+ if usedefault:
+ socketmap = asyncore.socket_map
+ try:
+ asyncore.socket_map = testmap
+ asyncore.close_all()
+ finally:
+ testmap, asyncore.socket_map = asyncore.socket_map, socketmap
+ else:
+ asyncore.close_all(testmap)
+
+ self.assertEqual(len(testmap), 0)
+
+ for c in l:
+ self.assertEqual(c.socket.closed, True)
+
+ def test_compact_traceback(self):
+ try:
+ raise Exception("I don't like spam!")
+ except:
+ real_t, real_v, real_tb = sys.exc_info()
+ r = asyncore.compact_traceback()
+ else:
+ self.fail("Expected exception")
+
+ (f, function, line), t, v, info = r
+ self.assertEqual(os.path.split(f)[-1], 'test_asyncore.py')
+ self.assertEqual(function, 'test_compact_traceback')
+ self.assertEqual(t, real_t)
+ self.assertEqual(v, real_v)
+ self.assertEqual(info, '[%s|%s|%s]' % (f, function, line))
+
+
+class DispatcherTests(unittest.TestCase):
+ def setUp(self):
+ pass
+
+ def tearDown(self):
+ asyncore.close_all()
+
+ def test_basic(self):
+ d = asyncore.dispatcher()
+ self.assertEqual(d.readable(), True)
+ self.assertEqual(d.writable(), True)
+
+ def test_repr(self):
+ d = asyncore.dispatcher()
+ self.assertEqual(repr(d), '<asyncore.dispatcher at %#x>' % id(d))
+
+ def test_log(self):
+ d = asyncore.dispatcher()
+
+ # capture output of dispatcher.log() (to stderr)
+ fp = StringIO()
+ stderr = sys.stderr
+ l1 = "Lovely spam! Wonderful spam!"
+ l2 = "I don't like spam!"
+ try:
+ sys.stderr = fp
+ d.log(l1)
+ d.log(l2)
+ finally:
+ sys.stderr = stderr
+
+ lines = fp.getvalue().splitlines()
+ self.assertEqual(lines, ['log: %s' % l1, 'log: %s' % l2])
+
+ def test_log_info(self):
+ d = asyncore.dispatcher()
+
+ # capture output of dispatcher.log_info() (to stdout via print)
+ fp = StringIO()
+ stdout = sys.stdout
+ l1 = "Have you got anything without spam?"
+ l2 = "Why can't she have egg bacon spam and sausage?"
+ l3 = "THAT'S got spam in it!"
+ try:
+ sys.stdout = fp
+ d.log_info(l1, 'EGGS')
+ d.log_info(l2)
+ d.log_info(l3, 'SPAM')
+ finally:
+ sys.stdout = stdout
+
+ lines = fp.getvalue().splitlines()
+ expected = ['EGGS: %s' % l1, 'info: %s' % l2, 'SPAM: %s' % l3]
+
+ self.assertEqual(lines, expected)
+
+ def test_unhandled(self):
+ d = asyncore.dispatcher()
+ d.ignore_log_types = ()
+
+ # capture output of dispatcher.log_info() (to stdout via print)
+ fp = StringIO()
+ stdout = sys.stdout
+ try:
+ sys.stdout = fp
+ d.handle_expt()
+ d.handle_read()
+ d.handle_write()
+ d.handle_connect()
+ d.handle_accept()
+ finally:
+ sys.stdout = stdout
+
+ lines = fp.getvalue().splitlines()
+ expected = ['warning: unhandled incoming priority event',
+ 'warning: unhandled read event',
+ 'warning: unhandled write event',
+ 'warning: unhandled connect event',
+ 'warning: unhandled accept event']
+ self.assertEqual(lines, expected)
+
+ def test_issue_8594(self):
+ # XXX - this test is supposed to be removed in next major Python
+ # version
+ d = asyncore.dispatcher(socket.socket())
+ # make sure the error message no longer refers to the socket
+ # object but the dispatcher instance instead
+ self.assertRaisesRegexp(AttributeError, 'dispatcher instance',
+ getattr, d, 'foo')
+ # cheap inheritance with the underlying socket is supposed
+ # to still work but a DeprecationWarning is expected
+ with warnings.catch_warnings(record=True) as w:
+ warnings.simplefilter("always")
+ family = d.family
+ self.assertEqual(family, socket.AF_INET)
+ self.assertEqual(len(w), 1)
+ self.assertTrue(issubclass(w[0].category, DeprecationWarning))
+
+ def test_strerror(self):
+ # refers to bug #8573
+ err = asyncore._strerror(errno.EPERM)
+ if hasattr(os, 'strerror'):
+ self.assertEqual(err, os.strerror(errno.EPERM))
+ err = asyncore._strerror(-1)
+ self.assertTrue(err != "")
+
+
+class dispatcherwithsend_noread(asyncore.dispatcher_with_send):
+ def readable(self):
+ return False
+
+ def handle_connect(self):
+ pass
+
+class DispatcherWithSendTests(unittest.TestCase):
+ usepoll = False
+
+ def setUp(self):
+ pass
+
+ def tearDown(self):
+ asyncore.close_all()
+
+ @unittest.skipUnless(threading, 'Threading required for this test.')
+ @test_support.reap_threads
+ def test_send(self):
+ evt = threading.Event()
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ sock.settimeout(3)
+ port = test_support.bind_port(sock)
+
+ cap = StringIO()
+ args = (evt, cap, sock)
+ t = threading.Thread(target=capture_server, args=args)
+ t.start()
+ try:
+ # wait a little longer for the server to initialize (it sometimes
+ # refuses connections on slow machines without this wait)
+ time.sleep(0.2)
+
+ data = "Suppose there isn't a 16-ton weight?"
+ d = dispatcherwithsend_noread()
+ d.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ d.connect((HOST, port))
+
+ # give time for socket to connect
+ time.sleep(0.1)
+
+ d.send(data)
+ d.send(data)
+ d.send('\n')
+
+ n = 1000
+ while d.out_buffer and n > 0:
+ asyncore.poll()
+ n -= 1
+
+ evt.wait()
+
+ self.assertEqual(cap.getvalue(), data*2)
+ finally:
+ t.join()
+
+
+class DispatcherWithSendTests_UsePoll(DispatcherWithSendTests):
+ usepoll = True
+
+ at unittest.skipUnless(hasattr(asyncore, 'file_wrapper'),
+ 'asyncore.file_wrapper required')
+class FileWrapperTest(unittest.TestCase):
+ def setUp(self):
+ self.d = "It's not dead, it's sleeping!"
+ with file(TESTFN, 'w') as h:
+ h.write(self.d)
+
+ def tearDown(self):
+ unlink(TESTFN)
+
+ def test_recv(self):
+ fd = os.open(TESTFN, os.O_RDONLY)
+ w = asyncore.file_wrapper(fd)
+ os.close(fd)
+
+ self.assertNotEqual(w.fd, fd)
+ self.assertNotEqual(w.fileno(), fd)
+ self.assertEqual(w.recv(13), "It's not dead")
+ self.assertEqual(w.read(6), ", it's")
+ w.close()
+ self.assertRaises(OSError, w.read, 1)
+
+
+ def test_send(self):
+ d1 = "Come again?"
+ d2 = "I want to buy some cheese."
+ fd = os.open(TESTFN, os.O_WRONLY | os.O_APPEND)
+ w = asyncore.file_wrapper(fd)
+ os.close(fd)
+
+ w.write(d1)
+ w.send(d2)
+ w.close()
+ self.assertEqual(file(TESTFN).read(), self.d + d1 + d2)
+
+ @unittest.skipUnless(hasattr(asyncore, 'file_dispatcher'),
+ 'asyncore.file_dispatcher required')
+ def test_dispatcher(self):
+ fd = os.open(TESTFN, os.O_RDONLY)
+ data = []
+ class FileDispatcher(asyncore.file_dispatcher):
+ def handle_read(self):
+ data.append(self.recv(29))
+ s = FileDispatcher(fd)
+ os.close(fd)
+ asyncore.loop(timeout=0.01, use_poll=True, count=2)
+ self.assertEqual(b"".join(data), self.d)
+
+
+class BaseTestHandler(asyncore.dispatcher):
+
+ def __init__(self, sock=None):
+ asyncore.dispatcher.__init__(self, sock)
+ self.flag = False
+
+ def handle_accept(self):
+ raise Exception("handle_accept not supposed to be called")
+
+ def handle_connect(self):
+ raise Exception("handle_connect not supposed to be called")
+
+ def handle_expt(self):
+ raise Exception("handle_expt not supposed to be called")
+
+ def handle_close(self):
+ raise Exception("handle_close not supposed to be called")
+
+ def handle_error(self):
+ raise
+
+
+class TCPServer(asyncore.dispatcher):
+ """A server which listens on an address and dispatches the
+ connection to a handler.
+ """
+
+ def __init__(self, handler=BaseTestHandler, host=HOST, port=0):
+ asyncore.dispatcher.__init__(self)
+ self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.set_reuse_addr()
+ self.bind((host, port))
+ self.listen(5)
+ self.handler = handler
+
+ @property
+ def address(self):
+ return self.socket.getsockname()[:2]
+
+ def handle_accept(self):
+ sock, addr = self.accept()
+ self.handler(sock)
+
+ def handle_error(self):
+ raise
+
+
+class BaseClient(BaseTestHandler):
+
+ def __init__(self, address):
+ BaseTestHandler.__init__(self)
+ self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.connect(address)
+
+ def handle_connect(self):
+ pass
+
+
+class BaseTestAPI(unittest.TestCase):
+
+ def tearDown(self):
+ asyncore.close_all()
+
+ def loop_waiting_for_flag(self, instance, timeout=5):
+ timeout = float(timeout) / 100
+ count = 100
+ while asyncore.socket_map and count > 0:
+ asyncore.loop(timeout=0.01, count=1, use_poll=self.use_poll)
+ if instance.flag:
+ return
+ count -= 1
+ time.sleep(timeout)
+ self.fail("flag not set")
+
+ def test_handle_connect(self):
+ # make sure handle_connect is called on connect()
+
+ class TestClient(BaseClient):
+ def handle_connect(self):
+ self.flag = True
+
+ server = TCPServer()
+ client = TestClient(server.address)
+ self.loop_waiting_for_flag(client)
+
+ def test_handle_accept(self):
+ # make sure handle_accept() is called when a client connects
+
+ class TestListener(BaseTestHandler):
+
+ def __init__(self):
+ BaseTestHandler.__init__(self)
+ self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.bind((HOST, 0))
+ self.listen(5)
+ self.address = self.socket.getsockname()[:2]
+
+ def handle_accept(self):
+ self.flag = True
+
+ server = TestListener()
+ client = BaseClient(server.address)
+ self.loop_waiting_for_flag(server)
+
+ def test_handle_read(self):
+ # make sure handle_read is called on data received
+
+ class TestClient(BaseClient):
+ def handle_read(self):
+ self.flag = True
+
+ class TestHandler(BaseTestHandler):
+ def __init__(self, conn):
+ BaseTestHandler.__init__(self, conn)
+ self.send('x' * 1024)
+
+ server = TCPServer(TestHandler)
+ client = TestClient(server.address)
+ self.loop_waiting_for_flag(client)
+
+ def test_handle_write(self):
+ # make sure handle_write is called
+
+ class TestClient(BaseClient):
+ def handle_write(self):
+ self.flag = True
+
+ server = TCPServer()
+ client = TestClient(server.address)
+ self.loop_waiting_for_flag(client)
+
+ def test_handle_close(self):
+ # make sure handle_close is called when the other end closes
+ # the connection
+
+ class TestClient(BaseClient):
+
+ def handle_read(self):
+ # in order to make handle_close be called we are supposed
+ # to make at least one recv() call
+ self.recv(1024)
+
+ def handle_close(self):
+ self.flag = True
+ self.close()
+
+ class TestHandler(BaseTestHandler):
+ def __init__(self, conn):
+ BaseTestHandler.__init__(self, conn)
+ self.close()
+
+ server = TCPServer(TestHandler)
+ client = TestClient(server.address)
+ self.loop_waiting_for_flag(client)
+
+ @unittest.skipIf(sys.platform.startswith("sunos"),
+ "OOB support is broken on Solaris")
+ def test_handle_expt(self):
+ # Make sure handle_expt is called on OOB data received.
+ # Note: this might fail on some platforms as OOB data is
+ # tenuously supported and rarely used.
+
+ class TestClient(BaseClient):
+ def handle_expt(self):
+ self.flag = True
+
+ class TestHandler(BaseTestHandler):
+ def __init__(self, conn):
+ BaseTestHandler.__init__(self, conn)
+ self.socket.send(chr(244), socket.MSG_OOB)
+
+ server = TCPServer(TestHandler)
+ client = TestClient(server.address)
+ self.loop_waiting_for_flag(client)
+
+ def test_handle_error(self):
+
+ class TestClient(BaseClient):
+ def handle_write(self):
+ 1.0 / 0
+ def handle_error(self):
+ self.flag = True
+ try:
+ raise
+ except ZeroDivisionError:
+ pass
+ else:
+ raise Exception("exception not raised")
+
+ server = TCPServer()
+ client = TestClient(server.address)
+ self.loop_waiting_for_flag(client)
+
+ def test_connection_attributes(self):
+ server = TCPServer()
+ client = BaseClient(server.address)
+
+ # we start disconnected
+ self.assertFalse(server.connected)
+ self.assertTrue(server.accepting)
+ # this can't be taken for granted across all platforms
+ #self.assertFalse(client.connected)
+ self.assertFalse(client.accepting)
+
+ # execute some loops so that client connects to server
+ asyncore.loop(timeout=0.01, use_poll=self.use_poll, count=100)
+ self.assertFalse(server.connected)
+ self.assertTrue(server.accepting)
+ self.assertTrue(client.connected)
+ self.assertFalse(client.accepting)
+
+ # disconnect the client
+ client.close()
+ self.assertFalse(server.connected)
+ self.assertTrue(server.accepting)
+ self.assertFalse(client.connected)
+ self.assertFalse(client.accepting)
+
+ # stop serving
+ server.close()
+ self.assertFalse(server.connected)
+ self.assertFalse(server.accepting)
+
+ def test_create_socket(self):
+ s = asyncore.dispatcher()
+ s.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ self.assertEqual(s.socket.family, socket.AF_INET)
+ self.assertEqual(s.socket.type, socket.SOCK_STREAM)
+
+ def test_bind(self):
+ s1 = asyncore.dispatcher()
+ s1.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ s1.bind((HOST, 0))
+ s1.listen(5)
+ port = s1.socket.getsockname()[1]
+
+ s2 = asyncore.dispatcher()
+ s2.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ # EADDRINUSE indicates the socket was correctly bound
+ self.assertRaises(socket.error, s2.bind, (HOST, port))
+
+ def test_set_reuse_addr(self):
+ sock = socket.socket()
+ try:
+ sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+ except socket.error:
+ unittest.skip("SO_REUSEADDR not supported on this platform")
+ else:
+ # if SO_REUSEADDR succeeded for sock we expect asyncore
+ # to do the same
+ s = asyncore.dispatcher(socket.socket())
+ self.assertFalse(s.socket.getsockopt(socket.SOL_SOCKET,
+ socket.SO_REUSEADDR))
+ s.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+ s.set_reuse_addr()
+ self.assertTrue(s.socket.getsockopt(socket.SOL_SOCKET,
+ socket.SO_REUSEADDR))
+ finally:
+ sock.close()
+
+ @unittest.skipUnless(threading, 'Threading required for this test.')
+ @test_support.reap_threads
+ def test_quick_connect(self):
+ # see: http://bugs.python.org/issue10340
+ server = TCPServer()
+ t = threading.Thread(target=lambda: asyncore.loop(timeout=0.1, count=500))
+ t.start()
+
+ for x in xrange(20):
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ s.settimeout(.2)
+ s.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER,
+ struct.pack('ii', 1, 0))
+ try:
+ s.connect(server.address)
+ except socket.error:
+ pass
+ finally:
+ s.close()
+
+
+class TestAPI_UseSelect(BaseTestAPI):
+ use_poll = False
+
+ at unittest.skipUnless(hasattr(select, 'poll'), 'select.poll required')
+class TestAPI_UsePoll(BaseTestAPI):
+ use_poll = True
+
+
+def test_main():
+ tests = [HelperFunctionTests, DispatcherTests, DispatcherWithSendTests,
+ DispatcherWithSendTests_UsePoll, TestAPI_UseSelect,
+ TestAPI_UsePoll, FileWrapperTest]
+ run_unittest(*tests)
+
+if __name__ == "__main__":
+ test_main()
--
Repository URL: http://hg.python.org/jython
More information about the Jython-checkins
mailing list