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