mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-12-04 00:30:19 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

bpo-44011: New asyncio ssl implementation (#17975)

Browse source
This commit is contained in:
Andrew Svetlov 2021-05-03 00:34:15 +03:00 • committed by GitHub
parent c96cc089f6
commit 5fb06edbbb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
12 changed files with 2474 additions and 527 deletions
Download patch file Download diff file Expand all files Collapse all files

38
Lib/test/test_asyncio/test_selector_events.py
View file

@ -70,44 +70,6 @@ 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):