diff options
author | Pablo Galindo <Pablogsal@gmail.com> | 2021-05-03 16:21:59 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-05-03 16:21:59 +0100 |
commit | 7719953b30430b351ba0f153c2b51b16cc68ee36 (patch) | |
tree | 8014086b85a13ed79d45e29ab74a9a9f5c9c68eb /Lib/test/test_asyncio/test_selector_events.py | |
parent | 39494285e15dc2d291ec13de5045b930eaf0a3db (diff) | |
download | cpython-git-7719953b30430b351ba0f153c2b51b16cc68ee36.tar.gz |
bpo-44011: Revert "New asyncio ssl implementation (GH-17975)" (GH-25848)
This reverts commit 5fb06edbbb769561e245d0fe13002bab50e2ae60 and all
subsequent dependent commits.
Diffstat (limited to 'Lib/test/test_asyncio/test_selector_events.py')
-rw-r--r-- | Lib/test/test_asyncio/test_selector_events.py | 38 |
1 files changed, 38 insertions, 0 deletions
diff --git a/Lib/test/test_asyncio/test_selector_events.py b/Lib/test/test_asyncio/test_selector_events.py index 349e4f2dca..1613c753c2 100644 --- a/Lib/test/test_asyncio/test_selector_events.py +++ b/Lib/test/test_asyncio/test_selector_events.py @@ -70,6 +70,44 @@ class BaseSelectorEventLoopTests(test_utils.TestCase): close_transport(transport) + @unittest.skipIf(ssl is None, 'No ssl module') + def test_make_ssl_transport(self): + m = mock.Mock() + self.loop._add_reader = mock.Mock() + self.loop._add_reader._is_coroutine = False + self.loop._add_writer = mock.Mock() + self.loop._remove_reader = mock.Mock() + self.loop._remove_writer = mock.Mock() + waiter = self.loop.create_future() + with test_utils.disable_logger(): + transport = self.loop._make_ssl_transport( + m, asyncio.Protocol(), m, waiter) + + with self.assertRaisesRegex(RuntimeError, + r'SSL transport.*not.*initialized'): + transport.is_reading() + + # execute the handshake while the logger is disabled + # to ignore SSL handshake failure + test_utils.run_briefly(self.loop) + + self.assertTrue(transport.is_reading()) + transport.pause_reading() + transport.pause_reading() + self.assertFalse(transport.is_reading()) + transport.resume_reading() + transport.resume_reading() + self.assertTrue(transport.is_reading()) + + # Sanity check + class_name = transport.__class__.__name__ + self.assertIn("ssl", class_name.lower()) + self.assertIn("transport", class_name.lower()) + + transport.close() + # execute pending callbacks to close the socket transport + test_utils.run_briefly(self.loop) + @mock.patch('asyncio.selector_events.ssl', None) @mock.patch('asyncio.sslproto.ssl', None) def test_make_ssl_transport_without_ssl_error(self): |