diff options
Diffstat (limited to 'Lib/test')
| -rw-r--r-- | Lib/test/test_asyncio/test_events.py | 68 | ||||
| -rw-r--r-- | Lib/test/test_asyncio/test_selector_events.py | 363 |
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): |