3 from unittest import main
6 from eventlet import greenio
8 from eventlet.green import ssl
12 LimitedTestCase, certificate_file, private_key_file, check_idle_cpu_usage,
17 def listen_ssl_socket(address=('127.0.0.1', 0)):
18 sock = ssl.wrap_socket(
19 socket.socket(), private_key_file, certificate_file, server_side=True)
26 class SSLTest(LimitedTestCase):
28 # disabling socket.ssl warnings because we're testing it here
29 warnings.filterwarnings(
31 message='.*socket.ssl.*',
32 category=DeprecationWarning)
34 super(SSLTest, self).setUp()
37 def test_duplex_response(self):
39 sock, addr = listener.accept()
41 sock.write(b'response')
43 sock = listen_ssl_socket()
45 server_coro = eventlet.spawn(serve, sock)
47 client = ssl.wrap_socket(
48 eventlet.connect(('127.0.0.1', sock.getsockname()[1])))
49 client.write(b'line 1\r\nline 2\r\n\r\n')
50 self.assertEqual(client.read(8192), b'response')
54 def test_ssl_close(self):
56 sock, addr = listener.accept()
59 self.assertEqual(b"", sock.read(8192))
60 except greenio.SSL.ZeroReturnError:
63 sock = listen_ssl_socket()
65 server_coro = eventlet.spawn(serve, sock)
67 raw_client = eventlet.connect(('127.0.0.1', sock.getsockname()[1]))
68 client = ssl.wrap_socket(raw_client)
70 greenio.shutdown_safe(client)
75 def test_ssl_connect(self):
77 sock, addr = listener.accept()
79 sock = listen_ssl_socket()
80 server_coro = eventlet.spawn(serve, sock)
82 raw_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
83 ssl_client = ssl.wrap_socket(raw_client)
84 ssl_client.connect(('127.0.0.1', sock.getsockname()[1]))
85 ssl_client.write(b'abc')
86 greenio.shutdown_safe(ssl_client)
91 def test_ssl_unwrap(self):
93 sock, addr = listener.accept()
94 self.assertEqual(sock.recv(6), b'before')
95 sock_ssl = ssl.wrap_socket(sock, private_key_file, certificate_file,
97 sock_ssl.do_handshake()
98 self.assertEqual(sock_ssl.read(6), b'during')
99 sock2 = sock_ssl.unwrap()
100 self.assertEqual(sock2.recv(5), b'after')
103 listener = eventlet.listen(('127.0.0.1', 0))
104 server_coro = eventlet.spawn(serve)
105 client = eventlet.connect((listener.getsockname()))
106 client.send(b'before')
107 client_ssl = ssl.wrap_socket(client)
108 client_ssl.do_handshake()
109 client_ssl.write(b'during')
110 client2 = client_ssl.unwrap()
111 client2.send(b'after')
115 def test_sendall_cpu_usage(self):
116 """SSL socket.sendall() busy loop
118 https://bitbucket.org/eventlet/eventlet/issue/134/greenssl-performance-issues
120 Idea of this test is to check that GreenSSLSocket.sendall() does not busy loop
121 retrying .send() calls, but instead trampolines until socket is writeable.
123 BUFFER_SIZE and SENDALL_SIZE are magic numbers inferred through trial and error.
125 # Time limit resistant to busy loops
128 stage_1 = eventlet.event.Event()
130 SENDALL_SIZE = 100000
133 conn, _ = listener.accept()
134 conn.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, BUFFER_SIZE)
135 self.assertEqual(conn.read(8), b'request')
136 conn.write(b'response')
139 conn.sendall(b'x' * SENDALL_SIZE)
141 server_sock = listen_ssl_socket()
142 server_coro = eventlet.spawn(serve, server_sock)
144 client_sock = eventlet.connect(server_sock.getsockname())
145 client_sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, BUFFER_SIZE)
146 client = ssl.wrap_socket(client_sock)
147 client.write(b'request')
148 self.assertEqual(client.read(8), b'response')
151 check_idle_cpu_usage(0.2, 0.1)
155 def test_greensslobject(self):
157 sock, addr = listener.accept()
158 sock.write(b'content')
159 greenio.shutdown_safe(sock)
161 listener = listen_ssl_socket(('', 0))
162 eventlet.spawn(serve, listener)
163 client = ssl.wrap_socket(
164 eventlet.connect(('localhost', listener.getsockname()[1])))
165 self.assertEqual(client.read(1024), b'content')
166 self.assertEqual(client.read(1024), b'')
169 def test_regression_gh_17(self):
171 sock, addr = listener.accept()
173 # to simulate condition mentioned in GH-17
175 sock.sendall(b'some data')
176 greenio.shutdown_safe(sock)
179 listener = listen_ssl_socket(('', 0))
180 eventlet.spawn(serve, listener)
181 ssl.wrap_socket(eventlet.connect(('localhost', listener.getsockname()[1])))
183 if __name__ == '__main__':