86 lines
3.1 KiB
Python
86 lines
3.1 KiB
Python
from tests import skipped, LimitedTestCase, skip_unless
|
|
from unittest import main
|
|
from eventlet import api, util, coros, greenio
|
|
import socket
|
|
import os
|
|
|
|
certificate_file = os.path.join(os.path.dirname(__file__), 'test_server.crt')
|
|
private_key_file = os.path.join(os.path.dirname(__file__), 'test_server.key')
|
|
|
|
class SSLTest(LimitedTestCase):
|
|
def test_duplex_response(self):
|
|
def serve(listener):
|
|
sock, addr = listener.accept()
|
|
stuff = sock.read(8192)
|
|
sock.write('response')
|
|
|
|
sock = api.ssl_listener(('127.0.0.1', 0), certificate_file, private_key_file)
|
|
server_coro = coros.execute(serve, sock)
|
|
|
|
client = util.wrap_ssl(api.connect_tcp(('127.0.0.1', sock.getsockname()[1])))
|
|
client.write('line 1\r\nline 2\r\n\r\n')
|
|
self.assertEquals(client.read(8192), 'response')
|
|
server_coro.wait()
|
|
|
|
def test_ssl_close(self):
|
|
def serve(listener):
|
|
sock, addr = listener.accept()
|
|
stuff = sock.read(8192)
|
|
try:
|
|
self.assertEquals("", sock.read(8192))
|
|
except greenio.SSL.ZeroReturnError:
|
|
pass
|
|
|
|
sock = api.ssl_listener(('127.0.0.1', 0), certificate_file, private_key_file)
|
|
server_coro = coros.execute(serve, sock)
|
|
|
|
raw_client = api.connect_tcp(('127.0.0.1', sock.getsockname()[1]))
|
|
client = util.wrap_ssl(raw_client)
|
|
client.write('X')
|
|
greenio.shutdown_safe(client)
|
|
client.close()
|
|
server_coro.wait()
|
|
|
|
def test_ssl_connect(self):
|
|
def serve(listener):
|
|
sock, addr = listener.accept()
|
|
stuff = sock.read(8192)
|
|
sock = api.ssl_listener(('127.0.0.1', 0), certificate_file, private_key_file)
|
|
server_coro = coros.execute(serve, sock)
|
|
|
|
raw_client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
ssl_client = util.wrap_ssl(raw_client)
|
|
ssl_client.connect(('127.0.0.1', sock.getsockname()[1]))
|
|
ssl_client.write('abc')
|
|
greenio.shutdown_safe(ssl_client)
|
|
ssl_client.close()
|
|
server_coro.wait()
|
|
|
|
|
|
class SocketSSLTest(LimitedTestCase):
|
|
@skip_unless(hasattr(socket, 'ssl'))
|
|
def test_greensslobject(self):
|
|
import warnings
|
|
# disabling socket.ssl warnings because we're testing it here
|
|
warnings.filterwarnings(action = 'ignore',
|
|
message='.*socket.ssl.*',
|
|
category=DeprecationWarning)
|
|
|
|
def serve(listener):
|
|
sock, addr = listener.accept()
|
|
sock.write('content')
|
|
greenio.shutdown_safe(sock)
|
|
sock.close()
|
|
listener = api.ssl_listener(('', 0),
|
|
certificate_file,
|
|
private_key_file)
|
|
killer = api.spawn(serve, listener)
|
|
from eventlet.green.socket import ssl
|
|
client = ssl(api.connect_tcp(('localhost', listener.getsockname()[1])))
|
|
self.assertEquals(client.read(1024), 'content')
|
|
self.assertEquals(client.read(1024), '')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|