text: """
  AMAK: 20050515: This module is the test_select.py from cpython 2.4, ported to jython + unittest
  """
  
  try:
      object
  except NameError:
      class object: pass
  
  import errno
  import select
  import socket
  
  import os
  import sys
  import test_support
  import unittest
  
  class SelectWrapper:
  
      def __init__(self):
          self.read_fds = []
          self.write_fds = []
          self.oob_fds = []
          self.timeout = None
  
      def add_read_fd(self, fd):
          self.read_fds.append(fd)
  
      def add_write_fd(self, fd):
          self.write_fds.append(fd)
  
      def add_oob_fd(self, fd):
          self.oob_fds.append(fd)
  
      def set_timeout(self, timeout):
          self.timeout = timeout
  
  class PollWrapper:
  
      def __init__(self):
          self.timeout = None
          self.poll_object = select.poll()
  
      def add_read_fd(self, fd):
          self.poll_object.register(fd, select.POLL_IN)
  
      def add_write_fd(self, fd):
          self.poll_object.register(fd, select.POLL_OUT)
  
      def add_oob_fd(self, fd):
          self.poll_object.register(fd, select.POLL_PRI)
  
  class TestSelectInvalidParameters(unittest.TestCase):
  
      def testBadSelectSetTypes(self):
          # Test some known error conditions
          for bad_select_set in [None, 1,]:
              for pos in range(2): # OOB not supported on Java
                  args = [[], [], []]
                  args[pos] = bad_select_set
                  try:
                      timeout = 0 # Can't wait forever
                      rfd, wfd, xfd = select.select(args[0], args[1], args[2], timeout)
                  except (select.error, TypeError):
                      pass
                  except Exception, x:
                      self.fail("Selecting on '%s' raised wrong exception %s" % (str(bad_select_set), str(x)))
                  else:
                      self.fail("Selecting on '%s' should have raised TypeError" % str(bad_select_set))
  
      def testBadSelectableTypes(self):
          class Nope: pass
  
          class Almost1:
              def fileno(self):
                  return 'fileno'
  
          class Almost2:
              def fileno(self):
                  return 'fileno'
  
          # Test some known error conditions
          for bad_selectable in [None, 1, object(), Nope(), Almost1(), Almost2()]:
              try:
                  timeout = 0 # Can't wait forever
                  rfd, wfd, xfd = select.select([bad_selectable], [], [], timeout)
              except (TypeError, select.error), x:
                  pass
              else:
                  self.fail("Selecting on '%s' should have raised TypeError or select.error" % str(bad_selectable))
  
      def testInvalidTimeoutTypes(self):
          for invalid_timeout in ['not a number']:
              try:
                  rfd, wfd, xfd = select.select([], [], [], invalid_timeout)
              except TypeError:
                  pass
              else:
                  self.fail("Invalid timeout value '%s' should have raised TypeError" % invalid_timeout)
  
      def testInvalidTimeoutValues(self):
          for invalid_timeout in [-1]:
              try:
                  rfd, wfd, xfd = select.select([], [], [], invalid_timeout)
              except (ValueError, select.error):
                  pass
              else:
                  self.fail("Invalid timeout value '%s' should have raised ValueError or select.error" % invalid_timeout)
  
  class TestSelectClientSocket(unittest.TestCase):
  
      def testUnconnectedSocket(self):
          sockets = [socket.socket(socket.AF_INET, socket.SOCK_STREAM) for x in range(5)]
          for pos in range(2): # OOB not supported on Java
              args = [[], [], []]
              args[pos] = sockets
              timeout = 0 # Can't wait forever
              rfd, wfd, xfd = select.select(args[0], args[1], args[2], timeout)
              for s in sockets:
                  self.failIf(s in rfd)
                  self.failIf(s in wfd)
  
  class TestPollClientSocket(unittest.TestCase):
  
      def testEventConstants(self):
          for event_name in ['IN', 'OUT', 'PRI', 'ERR', 'HUP', 'NVAL', ]:
              self.failUnless(hasattr(select, 'POLL%s' % event_name))
  
      def testUnregisterRaisesKeyError(self):
          s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
          poll_object = select.poll()
          try:
              poll_object.unregister(s)
          except KeyError:
              pass
          else:
              self.fail("Unregistering socket that is not registered should have raised KeyError")
  
  #
  # using the test_socket thread based server/client management, for convenience.
  #
  
  import test_socket
  
  class ThreadedPollClientSocket(test_socket.ThreadedTCPSocketTest):
  
      def testSocketRegisteredBeforeConnected(self):
          self.cli_conn = self.serv.accept()
  
      def _testSocketRegisteredBeforeConnected(self):
          timeout = 1000 # milliseconds
          poll_object = select.poll()
          # Register the socket before it is connected
          poll_object.register(self.cli, select.POLLOUT)
          result_list = poll_object.poll(timeout)
          result_sockets = [r[0] for r in result_list]
          self.failIf(self.cli in result_sockets, "Unconnected client socket should not have been selectable")
          # Now connect the socket, but DO NOT register it again
          self.cli.setblocking(0)
          self.cli.connect( (test_socket.HOST, test_socket.PORT) )
          # Now poll again, to check that the poll object has recognised that the socket is now connected
          result_list = poll_object.poll(timeout)
          result_sockets = [r[0] for r in result_list]
          self.failUnless(self.cli in result_sockets, "Connected client socket should have been selectable")
  
      def testSocketMustBeNonBlocking(self):
          self.cli_conn = self.serv.accept()
  
      def _testSocketMustBeNonBlocking(self):
          self.cli.setblocking(1)
          self.cli.connect( (test_socket.HOST, test_socket.PORT) )
          timeout = 1000 # milliseconds
          poll_object = select.poll()
          try:
              poll_object.register(self.cli)
          except select.error, se:
              self.failUnlessEqual(se[0], errno.ESOCKISBLOCKING)
          except Exception, x:
              self.fail("Registering blocking socket should have raised select.error, not %s" % str(x))
          else:
              self.fail("Registering blocking socket should have raised select.error")
  
  class TestPipes(unittest.TestCase):
  
      verbose = 1
  
      def test(self):
          import sys
          from test.test_support import verbose
          if sys.platform[:3] in ('win', 'mac', 'os2', 'riscos'):
              if verbose:
                  print "Can't test select easily on", sys.platform
              return
          cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done'
          p = os.popen(cmd, 'r')
          for tout in (0, 1, 2, 4, 8, 16) + (None,)*10:
              if verbose:
                  print 'timeout =', tout
              rfd, wfd, xfd = select.select([p], [], [], tout)
              if (rfd, wfd, xfd) == ([], [], []):
                  continue
              if (rfd, wfd, xfd) == ([p], [], []):
                  line = p.readline()
                  if verbose:
                      print repr(line)
                  if not line:
                      if verbose:
                          print 'EOF'
                      break
                  continue
              self.fail('Unexpected return values from select(): %s' % str(rfd, wfd, xfd))
          p.close()
  
  def test_main():
      tests = [
          TestSelectInvalidParameters,
          TestSelectClientSocket,
          TestPollClientSocket,
          ThreadedPollClientSocket,
      ]
      if sys.platform[:4] != 'java':
          tests.append(TestPipes)
      suites = [unittest.makeSuite(klass, 'test') for klass in tests]
      test_support.run_suite(unittest.TestSuite(suites))
  
  if __name__ == "__main__":
      test_main()
  