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

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