summaryrefslogtreecommitdiffstats
path: root/Lib/test
diff options
context:
space:
mode:
authorVictor Stinner <victor.stinner@gmail.com>2015-01-13 23:19:09 (GMT)
committerVictor Stinner <victor.stinner@gmail.com>2015-01-13 23:19:09 (GMT)
commit231b404cb026649d4b7172e75ac394ef558efe60 (patch)
tree6d647c68ec4583834213355cc8e21cf67987b2c5 /Lib/test
parent9036e49ba1acdd49ed7aa86a228a2657ca72c336 (diff)
downloadcpython-231b404cb026649d4b7172e75ac394ef558efe60.zip
cpython-231b404cb026649d4b7172e75ac394ef558efe60.tar.gz
cpython-231b404cb026649d4b7172e75ac394ef558efe60.tar.bz2
Issue #22560: New SSL implementation based on ssl.MemoryBIO
The new SSL implementation is based on the new ssl.MemoryBIO which is only available on Python 3.5. On Python 3.4 and older, the legacy SSL implementation (using SSL_write, SSL_read, etc.) is used. The proactor event loop only supports the new implementation. The new asyncio.sslproto module adds _SSLPipe, SSLProtocol and _SSLProtocolTransport classes. _SSLPipe allows to "wrap" or "unwrap" a socket (switch between cleartext and SSL/TLS). Patch written by Antoine Pitrou. sslproto.py is based on gruvi/ssl.py of the gruvi project written by Geert Jansen. This change adds SSL support to ProactorEventLoop on Python 3.5 and newer! It becomes also possible to implement STARTTTLS: switch a cleartext socket to SSL.
Diffstat (limited to 'Lib/test')
-rw-r--r--Lib/test/test_asyncio/test_events.py58
-rw-r--r--Lib/test/test_asyncio/test_selector_events.py6
2 files changed, 53 insertions, 11 deletions
diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index af2da1f..a2c6dc9 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -650,6 +650,10 @@ class EventLoopTestsMixin:
*httpd.address)
self._test_create_ssl_connection(httpd, create_connection)
+ def test_legacy_create_ssl_connection(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_ssl_connection()
+
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_ssl_unix_connection(self):
@@ -666,6 +670,10 @@ class EventLoopTestsMixin:
self._test_create_ssl_connection(httpd, create_connection,
check_sockname)
+ def test_legacy_create_ssl_unix_connection(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_ssl_unix_connection()
+
def test_create_connection_local_addr(self):
with test_utils.run_test_server() as httpd:
port = support.find_unused_port()
@@ -826,6 +834,10 @@ class EventLoopTestsMixin:
# stop serving
server.close()
+ def test_legacy_create_server_ssl(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_server_ssl()
+
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_unix_server_ssl(self):
@@ -857,6 +869,10 @@ class EventLoopTestsMixin:
# stop serving
server.close()
+ def test_legacy_create_unix_server_ssl(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_unix_server_ssl()
+
@unittest.skipIf(ssl is None, 'No ssl module')
def test_create_server_ssl_verify_failed(self):
proto = MyProto(loop=self.loop)
@@ -881,6 +897,10 @@ class EventLoopTestsMixin:
self.assertIsNone(proto.transport)
server.close()
+ def test_legacy_create_server_ssl_verify_failed(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_server_ssl_verify_failed()
+
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_unix_server_ssl_verify_failed(self):
@@ -907,6 +927,10 @@ class EventLoopTestsMixin:
self.assertIsNone(proto.transport)
server.close()
+ def test_legacy_create_unix_server_ssl_verify_failed(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_unix_server_ssl_verify_failed()
+
@unittest.skipIf(ssl is None, 'No ssl module')
def test_create_server_ssl_match_failed(self):
proto = MyProto(loop=self.loop)
@@ -934,6 +958,10 @@ class EventLoopTestsMixin:
proto.transport.close()
server.close()
+ def test_legacy_create_server_ssl_match_failed(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_server_ssl_match_failed()
+
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_unix_server_ssl_verified(self):
@@ -958,6 +986,11 @@ class EventLoopTestsMixin:
proto.transport.close()
client.close()
server.close()
+ self.loop.run_until_complete(proto.done)
+
+ def test_legacy_create_unix_server_ssl_verified(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_unix_server_ssl_verified()
@unittest.skipIf(ssl is None, 'No ssl module')
def test_create_server_ssl_verified(self):
@@ -982,6 +1015,11 @@ class EventLoopTestsMixin:
proto.transport.close()
client.close()
server.close()
+ self.loop.run_until_complete(proto.done)
+
+ def test_legacy_create_server_ssl_verified(self):
+ with test_utils.force_legacy_ssl_support():
+ self.test_create_server_ssl_verified()
def test_create_server_sock(self):
proto = asyncio.Future(loop=self.loop)
@@ -1746,20 +1784,20 @@ if sys.platform == 'win32':
def create_event_loop(self):
return asyncio.ProactorEventLoop()
- def test_create_ssl_connection(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with SSL")
+ def test_legacy_create_ssl_connection(self):
+ raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
- def test_create_server_ssl(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with SSL")
+ def test_legacy_create_server_ssl(self):
+ raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
- def test_create_server_ssl_verify_failed(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with SSL")
+ def test_legacy_create_server_ssl_verify_failed(self):
+ raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
- def test_create_server_ssl_match_failed(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with SSL")
+ def test_legacy_create_server_ssl_match_failed(self):
+ raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
- def test_create_server_ssl_verified(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with SSL")
+ def test_legacy_create_server_ssl_verified(self):
+ raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
def test_reader_callback(self):
raise unittest.SkipTest("IocpEventLoop does not have add_reader()")
diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py
index ff114f8..360327a 100644
--- a/Lib/test/test_asyncio/test_selector_events.py
+++ b/Lib/test/test_asyncio/test_selector_events.py
@@ -59,9 +59,13 @@ class BaseSelectorEventLoopTests(test_utils.TestCase):
with test_utils.disable_logger():
transport = self.loop._make_ssl_transport(
m, asyncio.Protocol(), m, waiter)
- self.assertIsInstance(transport, _SelectorSslTransport)
+ # Sanity check
+ class_name = transport.__class__.__name__
+ self.assertIn("ssl", class_name.lower())
+ self.assertIn("transport", class_name.lower())
@mock.patch('asyncio.selector_events.ssl', None)
+ @mock.patch('asyncio.sslproto.ssl', None)
def test_make_ssl_transport_without_ssl_error(self):
m = mock.Mock()
self.loop.add_reader = mock.Mock()