eventlet/tests/test__socket_errors.py

   1 import unittest
   2 import socket as _original_sock
   3 from eventlet.green import socket
   4
   5
   6 class TestSocketErrors(unittest.TestCase):
   7     def test_connection_refused(self):
   8         # open and close a dummy server to find an unused port
   9         server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  10         server.bind(('127.0.0.1', 0))
  11         server.listen(1)
  12         port = server.getsockname()[1]
  13         server.close()
  14         del server
  15         s = socket.socket()
  16         try:
  17             s.connect(('127.0.0.1', port))
  18             self.fail("Shouldn't have connected")
  19         except socket.error as ex:
  20             code, text = ex.args
  21             assert code in [111, 61, 10061], (code, text)
  22             assert 'refused' in text.lower(), (code, text)
  23
  24     def test_timeout_real_socket(self):
  25         """ Test underlying socket behavior to ensure correspondence
  26             between green sockets and the underlying socket module. """
  27         return self.test_timeout(socket=_original_sock)
  28
  29     def test_timeout(self, socket=socket):
  30         """ Test that the socket timeout exception works correctly. """
  31         server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
  32         server.bind(('127.0.0.1', 0))
  33         server.listen(1)
  34         port = server.getsockname()[1]
  35
  36         s = socket.socket()
  37
  38         s.connect(('127.0.0.1', port))
  39
  40         cs, addr = server.accept()
  41         cs.settimeout(1)
  42         try:
  43             try:
  44                 cs.recv(1024)
  45                 self.fail("Should have timed out")
  46             except socket.timeout as ex:
  47                 assert hasattr(ex, 'args')
  48                 assert len(ex.args) == 1
  49                 assert ex.args[0] == 'timed out'
  50         finally:
  51             s.close()
  52             cs.close()
  53             server.close()
  54
  55 if __name__ == '__main__':
  56     unittest.main()