ยปCore Development>Code coverage>Lib/test/test_asyncio/test_sslproto.py

Python code coverage for Lib/test/test_asyncio/test_sslproto.py

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