bpo-32069: Drop legacy SSL transport (#4451)

* Drop legacy SSL transport

* Drop unused import

* Fix Windows tests

* Drop never executed on Python 3.4+ code

2 files changed, 0 insertions, 431 deletions

diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
index 5394ddf..1a8bc13 100644
--- a/Lib/test/test_asyncio/test_events.py
+++ b/Lib/test/test_asyncio/test_events.py
@@ -736,10 +736,6 @@ class EventLoopTestsMixin:
             self._test_create_ssl_connection(httpd, create_connection,
                                              peername=httpd.address)
 
-    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):
@@ -757,10 +753,6 @@ class EventLoopTestsMixin:
                                              check_sockname,
                                              peername=httpd.address)
 
-    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()
@@ -1061,10 +1053,6 @@ 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):
@@ -1096,10 +1084,6 @@ 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)
@@ -1129,10 +1113,6 @@ 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):
@@ -1163,11 +1143,6 @@ 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)
@@ -1196,10 +1171,6 @@ 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):
@@ -1226,10 +1197,6 @@ class EventLoopTestsMixin:
         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):
         proto = MyProto(loop=self.loop)
@@ -1259,10 +1226,6 @@ class EventLoopTestsMixin:
         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)
 
@@ -2159,37 +2122,6 @@ if sys.platform == 'win32':
         def create_event_loop(self):
             return asyncio.ProactorEventLoop()
 
-        if not sslproto._is_sslproto_available():
-            def test_create_ssl_connection(self):
-                raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
-            def test_create_server_ssl(self):
-                raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
-            def test_create_server_ssl_verify_failed(self):
-                raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
-            def test_create_server_ssl_match_failed(self):
-                raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
-            def test_create_server_ssl_verified(self):
-                raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
-        def test_legacy_create_ssl_connection(self):
-            raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
-        def test_legacy_create_server_ssl(self):
-            raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
-        def test_legacy_create_server_ssl_verify_failed(self):
-            raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
-        def test_legacy_create_server_ssl_match_failed(self):
-            raise unittest.SkipTest("IocpEventLoop incompatible with legacy 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 a3d118e..e67afcc 100644
--- a/Lib/test/test_asyncio/test_selector_events.py
+++ b/Lib/test/test_asyncio/test_selector_events.py
@@ -14,7 +14,6 @@ from asyncio import selectors
 from asyncio import test_utils
 from asyncio.selector_events import BaseSelectorEventLoop
 from asyncio.selector_events import _SelectorTransport
-from asyncio.selector_events import _SelectorSslTransport
 from asyncio.selector_events import _SelectorSocketTransport
 from asyncio.selector_events import _SelectorDatagramTransport
 
@@ -1178,368 +1177,6 @@ class SelectorSocketTransportTests(test_utils.TestCase):
         remove_writer.assert_called_with(self.sock_fd)
 
 
-@unittest.skipIf(ssl is None, 'No ssl module')
-class SelectorSslTransportTests(test_utils.TestCase):
-
-    def setUp(self):
-        super().setUp()
-        self.loop = self.new_test_loop()
-        self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
-        self.sock = mock.Mock(socket.socket)
-        self.sock.fileno.return_value = 7
-        self.sslsock = mock.Mock()
-        self.sslsock.fileno.return_value = 1
-        self.sslcontext = mock.Mock()
-        self.sslcontext.wrap_socket.return_value = self.sslsock
-
-    def ssl_transport(self, waiter=None, server_hostname=None):
-        transport = _SelectorSslTransport(self.loop, self.sock, self.protocol,
-                                          self.sslcontext, waiter=waiter,
-                                          server_hostname=server_hostname)
-        self.addCleanup(close_transport, transport)
-        return transport
-
-    def _make_one(self, create_waiter=None):
-        transport = self.ssl_transport()
-        self.sock.reset_mock()
-        self.sslsock.reset_mock()
-        self.sslcontext.reset_mock()
-        self.loop.reset_counters()
-        return transport
-
-    def test_on_handshake(self):
-        waiter = asyncio.Future(loop=self.loop)
-        tr = self.ssl_transport(waiter=waiter)
-        self.assertTrue(self.sslsock.do_handshake.called)
-        self.loop.assert_reader(1, tr._read_ready)
-        test_utils.run_briefly(self.loop)
-        self.assertIsNone(waiter.result())
-
-    def test_on_handshake_reader_retry(self):
-        self.loop.set_debug(False)
-        self.sslsock.do_handshake.side_effect = ssl.SSLWantReadError
-        transport = self.ssl_transport()
-        self.loop.assert_reader(1, transport._on_handshake, None)
-
-    def test_on_handshake_writer_retry(self):
-        self.loop.set_debug(False)
-        self.sslsock.do_handshake.side_effect = ssl.SSLWantWriteError
-        transport = self.ssl_transport()
-        self.loop.assert_writer(1, transport._on_handshake, None)
-
-    def test_on_handshake_exc(self):
-        exc = ValueError()
-        self.sslsock.do_handshake.side_effect = exc
-        with test_utils.disable_logger():
-            waiter = asyncio.Future(loop=self.loop)
-            self.ssl_transport(waiter=waiter)
-        self.assertTrue(waiter.done())
-        self.assertIs(exc, waiter.exception())
-        self.assertTrue(self.sslsock.close.called)
-
-    def test_on_handshake_base_exc(self):
-        waiter = asyncio.Future(loop=self.loop)
-        transport = self.ssl_transport(waiter=waiter)
-        exc = BaseException()
-        self.sslsock.do_handshake.side_effect = exc
-        with test_utils.disable_logger():
-            self.assertRaises(BaseException, transport._on_handshake, 0)
-        self.assertTrue(self.sslsock.close.called)
-        self.assertTrue(waiter.done())
-        self.assertIs(exc, waiter.exception())
-
-    def test_cancel_handshake(self):
-        # Python issue #23197: cancelling a handshake must not raise an
-        # exception or log an error, even if the handshake failed
-        waiter = asyncio.Future(loop=self.loop)
-        transport = self.ssl_transport(waiter=waiter)
-        waiter.cancel()
-        exc = ValueError()
-        self.sslsock.do_handshake.side_effect = exc
-        with test_utils.disable_logger():
-            transport._on_handshake(0)
-        transport.close()
-        test_utils.run_briefly(self.loop)
-
-    def test_pause_resume_reading(self):
-        tr = self._make_one()
-        self.assertFalse(tr._paused)
-        self.loop.assert_reader(1, tr._read_ready)
-        tr.pause_reading()
-        self.assertTrue(tr._paused)
-        self.assertFalse(1 in self.loop.readers)
-        tr.resume_reading()
-        self.assertFalse(tr._paused)
-        self.loop.assert_reader(1, tr._read_ready)
-        with self.assertRaises(RuntimeError):
-            tr.resume_reading()
-
-    def test_write(self):
-        transport = self._make_one()
-        transport.write(b'data')
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
-    def test_write_bytearray(self):
-        transport = self._make_one()
-        data = bytearray(b'data')
-        transport.write(data)
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-        self.assertEqual(data, bytearray(b'data'))  # Hasn't been mutated.
-        self.assertIsNot(data, transport._buffer)  # Hasn't been incorporated.
-
-    def test_write_memoryview(self):
-        transport = self._make_one()
-        data = memoryview(b'data')
-        transport.write(data)
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
-    def test_write_no_data(self):
-        transport = self._make_one()
-        transport._buffer.extend(b'data')
-        transport.write(b'')
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
-    def test_write_str(self):
-        transport = self._make_one()
-        self.assertRaises(TypeError, transport.write, 'str')
-
-    def test_write_closing(self):
-        transport = self._make_one()
-        transport.close()
-        self.assertEqual(transport._conn_lost, 1)
-        transport.write(b'data')
-        self.assertEqual(transport._conn_lost, 2)
-
-    @mock.patch('asyncio.selector_events.logger')
-    def test_write_exception(self, m_log):
-        transport = self._make_one()
-        transport._conn_lost = 1
-        transport.write(b'data')
-        self.assertEqual(transport._buffer, list_to_buffer())
-        transport.write(b'data')
-        transport.write(b'data')
-        transport.write(b'data')
-        transport.write(b'data')
-        m_log.warning.assert_called_with('socket.send() raised exception.')
-
-    def test_read_ready_recv(self):
-        self.sslsock.recv.return_value = b'data'
-        transport = self._make_one()
-        transport._read_ready()
-        self.assertTrue(self.sslsock.recv.called)
-        self.assertEqual((b'data',), self.protocol.data_received.call_args[0])
-
-    def test_read_ready_write_wants_read(self):
-        self.loop._add_writer = mock.Mock()
-        self.sslsock.recv.side_effect = BlockingIOError
-        transport = self._make_one()
-        transport._write_wants_read = True
-        transport._write_ready = mock.Mock()
-        transport._buffer.extend(b'data')
-        transport._read_ready()
-
-        self.assertFalse(transport._write_wants_read)
-        transport._write_ready.assert_called_with()
-        self.loop._add_writer.assert_called_with(
-            transport._sock_fd, transport._write_ready)
-
-    def test_read_ready_recv_eof(self):
-        self.sslsock.recv.return_value = b''
-        transport = self._make_one()
-        transport.close = mock.Mock()
-        transport._read_ready()
-        transport.close.assert_called_with()
-        self.protocol.eof_received.assert_called_with()
-
-    def test_read_ready_recv_conn_reset(self):
-        err = self.sslsock.recv.side_effect = ConnectionResetError()
-        transport = self._make_one()
-        transport._force_close = mock.Mock()
-        with test_utils.disable_logger():
-            transport._read_ready()
-        transport._force_close.assert_called_with(err)
-
-    def test_read_ready_recv_retry(self):
-        self.sslsock.recv.side_effect = ssl.SSLWantReadError
-        transport = self._make_one()
-        transport._read_ready()
-        self.assertTrue(self.sslsock.recv.called)
-        self.assertFalse(self.protocol.data_received.called)
-
-        self.sslsock.recv.side_effect = BlockingIOError
-        transport._read_ready()
-        self.assertFalse(self.protocol.data_received.called)
-
-        self.sslsock.recv.side_effect = InterruptedError
-        transport._read_ready()
-        self.assertFalse(self.protocol.data_received.called)
-
-    def test_read_ready_recv_write(self):
-        self.loop._remove_reader = mock.Mock()
-        self.loop._add_writer = mock.Mock()
-        self.sslsock.recv.side_effect = ssl.SSLWantWriteError
-        transport = self._make_one()
-        transport._read_ready()
-        self.assertFalse(self.protocol.data_received.called)
-        self.assertTrue(transport._read_wants_write)
-
-        self.loop._remove_reader.assert_called_with(transport._sock_fd)
-        self.loop._add_writer.assert_called_with(
-            transport._sock_fd, transport._write_ready)
-
-    def test_read_ready_recv_exc(self):
-        err = self.sslsock.recv.side_effect = OSError()
-        transport = self._make_one()
-        transport._fatal_error = mock.Mock()
-        transport._read_ready()
-        transport._fatal_error.assert_called_with(
-                                   err,
-                                   'Fatal read error on SSL transport')
-
-    def test_write_ready_send(self):
-        self.sslsock.send.return_value = 4
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data'])
-        transport._write_ready()
-        self.assertEqual(list_to_buffer(), transport._buffer)
-        self.assertTrue(self.sslsock.send.called)
-
-    def test_write_ready_send_none(self):
-        self.sslsock.send.return_value = 0
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data1', b'data2'])
-        transport._write_ready()
-        self.assertTrue(self.sslsock.send.called)
-        self.assertEqual(list_to_buffer([b'data1data2']), transport._buffer)
-
-    def test_write_ready_send_partial(self):
-        self.sslsock.send.return_value = 2
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data1', b'data2'])
-        transport._write_ready()
-        self.assertTrue(self.sslsock.send.called)
-        self.assertEqual(list_to_buffer([b'ta1data2']), transport._buffer)
-
-    def test_write_ready_send_closing_partial(self):
-        self.sslsock.send.return_value = 2
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data1', b'data2'])
-        transport._write_ready()
-        self.assertTrue(self.sslsock.send.called)
-        self.assertFalse(self.sslsock.close.called)
-
-    def test_write_ready_send_closing(self):
-        self.sslsock.send.return_value = 4
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data'])
-        transport.close()
-        transport._write_ready()
-        self.protocol.connection_lost.assert_called_with(None)
-
-    def test_write_ready_send_closing_empty_buffer(self):
-        self.sslsock.send.return_value = 4
-        call_soon = self.loop.call_soon = mock.Mock()
-        transport = self._make_one()
-        transport._buffer = list_to_buffer()
-        transport.close()
-        transport._write_ready()
-        call_soon.assert_called_with(transport._call_connection_lost, None)
-
-    def test_write_ready_send_retry(self):
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data'])
-
-        self.sslsock.send.side_effect = ssl.SSLWantWriteError
-        transport._write_ready()
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
-        self.sslsock.send.side_effect = BlockingIOError()
-        transport._write_ready()
-        self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
-    def test_write_ready_send_read(self):
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data'])
-
-        self.loop._remove_writer = mock.Mock()
-        self.sslsock.send.side_effect = ssl.SSLWantReadError
-        transport._write_ready()
-        self.assertFalse(self.protocol.data_received.called)
-        self.assertTrue(transport._write_wants_read)
-        self.loop._remove_writer.assert_called_with(transport._sock_fd)
-
-    def test_write_ready_send_exc(self):
-        err = self.sslsock.send.side_effect = OSError()
-
-        transport = self._make_one()
-        transport._buffer = list_to_buffer([b'data'])
-        transport._fatal_error = mock.Mock()
-        transport._write_ready()
-        transport._fatal_error.assert_called_with(
-                                   err,
-                                   'Fatal write error on SSL transport')
-        self.assertEqual(list_to_buffer(), transport._buffer)
-
-    def test_write_ready_read_wants_write(self):
-        self.loop._add_reader = mock.Mock()
-        self.sslsock.send.side_effect = BlockingIOError
-        transport = self._make_one()
-        transport._read_wants_write = True
-        transport._read_ready = mock.Mock()
-        transport._write_ready()
-
-        self.assertFalse(transport._read_wants_write)
-        transport._read_ready.assert_called_with()
-        self.loop._add_reader.assert_called_with(
-            transport._sock_fd, transport._read_ready)
-
-    def test_write_eof(self):
-        tr = self._make_one()
-        self.assertFalse(tr.can_write_eof())
-        self.assertRaises(NotImplementedError, tr.write_eof)
-
-    def check_close(self):
-        tr = self._make_one()
-        tr.close()
-
-        self.assertTrue(tr.is_closing())
-        self.assertEqual(1, self.loop.remove_reader_count[1])
-        self.assertEqual(tr._conn_lost, 1)
-
-        tr.close()
-        self.assertEqual(tr._conn_lost, 1)
-        self.assertEqual(1, self.loop.remove_reader_count[1])
-
-        test_utils.run_briefly(self.loop)
-
-    def test_close(self):
-        self.check_close()
-        self.assertTrue(self.protocol.connection_made.called)
-        self.assertTrue(self.protocol.connection_lost.called)
-
-    def test_close_not_connected(self):
-        self.sslsock.do_handshake.side_effect = ssl.SSLWantReadError
-        self.check_close()
-        self.assertFalse(self.protocol.connection_made.called)
-        self.assertFalse(self.protocol.connection_lost.called)
-
-    @unittest.skipIf(ssl is None, 'No SSL support')
-    def test_server_hostname(self):
-        self.ssl_transport(server_hostname='localhost')
-        self.sslcontext.wrap_socket.assert_called_with(
-            self.sock, do_handshake_on_connect=False, server_side=False,
-            server_hostname='localhost')
-
-
-class SelectorSslWithoutSslTransportTests(unittest.TestCase):
-
-    @mock.patch('asyncio.selector_events.ssl', None)
-    def test_ssl_transport_requires_ssl_module(self):
-        Mock = mock.Mock
-        with self.assertRaises(RuntimeError):
-            _SelectorSslTransport(Mock(), Mock(), Mock(), Mock())
-
-
 class SelectorDatagramTransportTests(test_utils.TestCase):
 
     def setUp(self):