tests/test__socket_errors.py

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