3 from unittest import main
6 from eventlet import util, greenio
8 from eventlet.green.socket 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 = util.wrap_ssl(socket.socket(), certificate_file,
19 private_key_file, 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 = util.wrap_ssl(eventlet.connect(('127.0.0.1', sock.getsockname()[1])))
48 client.write(b'line 1\r\nline 2\r\n\r\n')
49 self.assertEqual(client.read(8192), b'response')
53 def test_ssl_close(self):
55 sock, addr = listener.accept()
58 self.assertEqual(b"", sock.read(8192))
59 except greenio.SSL.ZeroReturnError:
62 sock = listen_ssl_socket()
64 server_coro = eventlet.spawn(serve, sock)
66 raw_client = eventlet.connect(('127.0.0.1', sock.getsockname()[1]))
67 client = util.wrap_ssl(raw_client)
69 greenio.shutdown_safe(client)
74 def test_ssl_connect(self):
76 sock, addr = listener.accept()
78 sock = listen_ssl_socket()
79 server_coro = eventlet.spawn(serve, sock)
81 raw_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
82 ssl_client = util.wrap_ssl(raw_client)
83 ssl_client.connect(('127.0.0.1', sock.getsockname()[1]))
84 ssl_client.write(b'abc')
85 greenio.shutdown_safe(ssl_client)
90 def test_ssl_unwrap(self):
92 sock, addr = listener.accept()
93 self.assertEqual(sock.recv(6), b'before')
94 sock_ssl = util.wrap_ssl(sock, certificate_file, private_key_file,
96 sock_ssl.do_handshake()
97 self.assertEqual(sock_ssl.read(6), b'during')
98 sock2 = sock_ssl.unwrap()
99 self.assertEqual(sock2.recv(5), b'after')
102 listener = eventlet.listen(('127.0.0.1', 0))
103 server_coro = eventlet.spawn(serve)
104 client = eventlet.connect((listener.getsockname()))
105 client.send(b'before')
106 client_ssl = util.wrap_ssl(client)
107 client_ssl.do_handshake()
108 client_ssl.write(b'during')
109 client2 = client_ssl.unwrap()
110 client2.send(b'after')
114 def test_sendall_cpu_usage(self):
115 """SSL socket.sendall() busy loop
117 https://bitbucket.org/eventlet/eventlet/issue/134/greenssl-performance-issues
119 Idea of this test is to check that GreenSSLSocket.sendall() does not busy loop
120 retrying .send() calls, but instead trampolines until socket is writeable.
122 BUFFER_SIZE and SENDALL_SIZE are magic numbers inferred through trial and error.
124 # Time limit resistant to busy loops
127 stage_1 = eventlet.event.Event()
129 SENDALL_SIZE = 100000
132 conn, _ = listener.accept()
133 conn.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, BUFFER_SIZE)
134 self.assertEqual(conn.read(8), b'request')
135 conn.write(b'response')
138 conn.sendall(b'x' * SENDALL_SIZE)
140 server_sock = listen_ssl_socket()
141 server_coro = eventlet.spawn(serve, server_sock)
143 client_sock = eventlet.connect(server_sock.getsockname())
144 client_sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, BUFFER_SIZE)
145 client = util.wrap_ssl(client_sock)
146 client.write(b'request')
147 self.assertEqual(client.read(8), b'response')
150 check_idle_cpu_usage(0.2, 0.1)
154 def test_greensslobject(self):
156 sock, addr = listener.accept()
157 sock.write(b'content')
158 greenio.shutdown_safe(sock)
160 listener = listen_ssl_socket(('', 0))
161 eventlet.spawn(serve, listener)
162 client = ssl(eventlet.connect(('localhost', listener.getsockname()[1])))
163 self.assertEqual(client.read(1024), b'content')
164 self.assertEqual(client.read(1024), b'')
167 def test_regression_gh_17(self):
169 sock, addr = listener.accept()
171 # to simulate condition mentioned in GH-17
173 sock.sendall(b'some data')
174 greenio.shutdown_safe(sock)
177 listener = listen_ssl_socket(('', 0))
178 eventlet.spawn(serve, listener)
179 ssl(eventlet.connect(('localhost', listener.getsockname()[1])))
181 if __name__ == '__main__':