2 import socket as _original_sock
3 from eventlet.green import socket
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))
12 port = server.getsockname()[1]
17 s.connect(('127.0.0.1', port))
18 self.fail("Shouldn't have connected")
19 except socket.error as ex:
21 assert code in [111, 61, 10061], (code, text)
22 assert 'refused' in text.lower(), (code, text)
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)
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))
34 port = server.getsockname()[1]
38 s.connect(('127.0.0.1', port))
40 cs, addr = server.accept()
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'
56 def test_create_connection_refused():
59 socket.create_connection(('127.0.0.1', 0))
60 except socket.error as ex:
62 assert errno in [111, 61, 10061], 'Expected socket.error ECONNREFUSED, got {0}'.format(errno)
65 if __name__ == '__main__':