1"""Tests for asyncio/sslproto.py.""" 2 3import logging 4import unittest 5from unittest import mock 6try: 7 import ssl 8except ImportError: 9 ssl = None 10 11import asyncio 12from asyncio import log 13from asyncio import sslproto 14from asyncio import test_utils 15 16 17@unittest.skipIf(ssl is None, 'No ssl module') 18class SslProtoHandshakeTests(test_utils.TestCase): 19 20 def setUp(self): 21 super().setUp() 22 self.loop = asyncio.new_event_loop() 23 self.set_event_loop(self.loop) 24 25 def ssl_protocol(self, waiter=None): 26 sslcontext = test_utils.dummy_ssl_context() 27 app_proto = asyncio.Protocol() 28 proto = sslproto.SSLProtocol(self.loop, app_proto, sslcontext, waiter) 29 self.assertIs(proto._app_transport.get_protocol(), app_proto) 30 self.addCleanup(proto._app_transport.close) 31 return proto 32 33 def connection_made(self, ssl_proto, do_handshake=None): 34 transport = mock.Mock() 35 sslpipe = mock.Mock() 36 sslpipe.shutdown.return_value = b'' 37 if do_handshake: 38 sslpipe.do_handshake.side_effect = do_handshake 39 else: 40 def mock_handshake(callback): 41 return [] 42 sslpipe.do_handshake.side_effect = mock_handshake 43 with mock.patch('asyncio.sslproto._SSLPipe', return_value=sslpipe): 44 ssl_proto.connection_made(transport) 45 46 def test_cancel_handshake(self): 47 # Python issue #23197: cancelling a handshake must not raise an 48 # exception or log an error, even if the handshake failed 49 waiter = asyncio.Future(loop=self.loop) 50 ssl_proto = self.ssl_protocol(waiter) 51 handshake_fut = asyncio.Future(loop=self.loop) 52 53 def do_handshake(callback): 54 exc = Exception() 55 callback(exc) 56 handshake_fut.set_result(None) 57 return [] 58 59 waiter.cancel() 60 self.connection_made(ssl_proto, do_handshake) 61 62 with test_utils.disable_logger(): 63 self.loop.run_until_complete(handshake_fut) 64 65 def test_eof_received_waiter(self): 66 waiter = asyncio.Future(loop=self.loop) 67 ssl_proto = self.ssl_protocol(waiter) 68 self.connection_made(ssl_proto) 69 ssl_proto.eof_received() 70 test_utils.run_briefly(self.loop) 71 self.assertIsInstance(waiter.exception(), ConnectionResetError) 72 73 def test_fatal_error_no_name_error(self): 74 # From issue #363. 75 # _fatal_error() generates a NameError if sslproto.py 76 # does not import base_events. 77 waiter = asyncio.Future(loop=self.loop) 78 ssl_proto = self.ssl_protocol(waiter) 79 # Temporarily turn off error logging so as not to spoil test output. 80 log_level = log.logger.getEffectiveLevel() 81 log.logger.setLevel(logging.FATAL) 82 try: 83 ssl_proto._fatal_error(None) 84 finally: 85 # Restore error logging. 86 log.logger.setLevel(log_level) 87 88 def test_connection_lost(self): 89 # From issue #472. 90 # yield from waiter hang if lost_connection was called. 91 waiter = asyncio.Future(loop=self.loop) 92 ssl_proto = self.ssl_protocol(waiter) 93 self.connection_made(ssl_proto) 94 ssl_proto.connection_lost(ConnectionAbortedError) 95 test_utils.run_briefly(self.loop) 96 self.assertIsInstance(waiter.exception(), ConnectionAbortedError) 97 98if __name__ == '__main__': 99 unittest.main() 100