2 import socket as _original_sock
3 from eventlet import api
4 from eventlet.green import socket
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))
13 port = server.getsockname()[1]
18 s.connect(('127.0.0.1', port))
19 self.fail("Shouldn't have connected")
20 except socket.error as ex:
22 assert code in [111, 61, 10061], (code, text)
23 assert 'refused' in text.lower(), (code, text)
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)
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))
35 port = server.getsockname()[1]
39 s.connect(('127.0.0.1', port))
41 cs, addr = server.accept()
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'
56 if __name__ == '__main__':