def _make_ssl_transport(self, rawsock, protocol, sslcontext, waiter=None,
*, server_side=False, server_hostname=None,
extra=None, server=None):
- if not sslproto._is_sslproto_available():
- raise NotImplementedError("Proactor event loop requires Python 3.5"
- " or newer (ssl.MemoryBIO) to support "
- "SSL")
-
ssl_protocol = sslproto.SSLProtocol(self, protocol, sslcontext, waiter,
server_side, server_hostname)
_ProactorSocketTransport(self, rawsock, ssl_protocol,
def _make_ssl_transport(self, rawsock, protocol, sslcontext, waiter=None,
*, server_side=False, server_hostname=None,
extra=None, server=None):
- if not sslproto._is_sslproto_available():
- return self._make_legacy_ssl_transport(
- rawsock, protocol, sslcontext, waiter,
- server_side=server_side, server_hostname=server_hostname,
- extra=extra, server=server)
-
ssl_protocol = sslproto.SSLProtocol(self, protocol, sslcontext, waiter,
server_side, server_hostname)
_SelectorSocketTransport(self, rawsock, ssl_protocol,
extra=extra, server=server)
return ssl_protocol._app_transport
- def _make_legacy_ssl_transport(self, rawsock, protocol, sslcontext,
- waiter, *,
- server_side=False, server_hostname=None,
- extra=None, server=None):
- # Use the legacy API: SSL_write, SSL_read, etc. The legacy API is used
- # on Python 3.4 and older, when ssl.MemoryBIO is not available.
- return _SelectorSslTransport(
- self, rawsock, protocol, sslcontext, waiter,
- server_side, server_hostname, extra, server)
-
def _make_datagram_transport(self, sock, protocol,
address=None, waiter=None, extra=None):
return _SelectorDatagramTransport(self, sock, protocol,
return True
-class _SelectorSslTransport(_SelectorTransport):
-
- _buffer_factory = bytearray
-
- def __init__(self, loop, rawsock, protocol, sslcontext, waiter=None,
- server_side=False, server_hostname=None,
- extra=None, server=None):
- if ssl is None:
- raise RuntimeError('stdlib ssl module not available')
-
- if not sslcontext:
- sslcontext = sslproto._create_transport_context(server_side, server_hostname)
-
- wrap_kwargs = {
- 'server_side': server_side,
- 'do_handshake_on_connect': False,
- }
- if server_hostname and not server_side:
- wrap_kwargs['server_hostname'] = server_hostname
- sslsock = sslcontext.wrap_socket(rawsock, **wrap_kwargs)
-
- super().__init__(loop, sslsock, protocol, extra, server)
- # the protocol connection is only made after the SSL handshake
- self._protocol_connected = False
-
- self._server_hostname = server_hostname
- self._waiter = waiter
- self._sslcontext = sslcontext
- self._paused = False
-
- # SSL-specific extra info. (peercert is set later)
- self._extra.update(sslcontext=sslcontext)
-
- if self._loop.get_debug():
- logger.debug("%r starts SSL handshake", self)
- start_time = self._loop.time()
- else:
- start_time = None
- self._on_handshake(start_time)
-
- def _wakeup_waiter(self, exc=None):
- if self._waiter is None:
- return
- if not self._waiter.cancelled():
- if exc is not None:
- self._waiter.set_exception(exc)
- else:
- self._waiter.set_result(None)
- self._waiter = None
-
- def _on_handshake(self, start_time):
- try:
- self._sock.do_handshake()
- except ssl.SSLWantReadError:
- self._loop._add_reader(self._sock_fd,
- self._on_handshake, start_time)
- return
- except ssl.SSLWantWriteError:
- self._loop._add_writer(self._sock_fd,
- self._on_handshake, start_time)
- return
- except BaseException as exc:
- if self._loop.get_debug():
- logger.warning("%r: SSL handshake failed",
- self, exc_info=True)
- self._loop._remove_reader(self._sock_fd)
- self._loop._remove_writer(self._sock_fd)
- self._sock.close()
- self._wakeup_waiter(exc)
- if isinstance(exc, Exception):
- return
- else:
- raise
-
- self._loop._remove_reader(self._sock_fd)
- self._loop._remove_writer(self._sock_fd)
-
- peercert = self._sock.getpeercert()
- if not hasattr(self._sslcontext, 'check_hostname'):
- # Verify hostname if requested, Python 3.4+ uses check_hostname
- # and checks the hostname in do_handshake()
- if (self._server_hostname and
- self._sslcontext.verify_mode != ssl.CERT_NONE):
- try:
- ssl.match_hostname(peercert, self._server_hostname)
- except Exception as exc:
- if self._loop.get_debug():
- logger.warning("%r: SSL handshake failed "
- "on matching the hostname",
- self, exc_info=True)
- self._sock.close()
- self._wakeup_waiter(exc)
- return
-
- # Add extra info that becomes available after handshake.
- self._extra.update(peercert=peercert,
- cipher=self._sock.cipher(),
- compression=self._sock.compression(),
- ssl_object=self._sock,
- )
-
- self._read_wants_write = False
- self._write_wants_read = False
- self._loop._add_reader(self._sock_fd, self._read_ready)
- self._protocol_connected = True
- self._loop.call_soon(self._protocol.connection_made, self)
- # only wake up the waiter when connection_made() has been called
- self._loop.call_soon(self._wakeup_waiter)
-
- if self._loop.get_debug():
- dt = self._loop.time() - start_time
- logger.debug("%r: SSL handshake took %.1f ms", self, dt * 1e3)
-
- def pause_reading(self):
- # XXX This is a bit icky, given the comment at the top of
- # _read_ready(). Is it possible to evoke a deadlock? I don't
- # know, although it doesn't look like it; write() will still
- # accept more data for the buffer and eventually the app will
- # call resume_reading() again, and things will flow again.
-
- if self._closing:
- raise RuntimeError('Cannot pause_reading() when closing')
- if self._paused:
- raise RuntimeError('Already paused')
- self._paused = True
- self._loop._remove_reader(self._sock_fd)
- if self._loop.get_debug():
- logger.debug("%r pauses reading", self)
-
- def resume_reading(self):
- if not self._paused:
- raise RuntimeError('Not paused')
- self._paused = False
- if self._closing:
- return
- self._loop._add_reader(self._sock_fd, self._read_ready)
- if self._loop.get_debug():
- logger.debug("%r resumes reading", self)
-
- def _read_ready(self):
- if self._conn_lost:
- return
- if self._write_wants_read:
- self._write_wants_read = False
- self._write_ready()
-
- if self._buffer:
- self._loop._add_writer(self._sock_fd, self._write_ready)
-
- try:
- data = self._sock.recv(self.max_size)
- except (BlockingIOError, InterruptedError, ssl.SSLWantReadError):
- pass
- except ssl.SSLWantWriteError:
- self._read_wants_write = True
- self._loop._remove_reader(self._sock_fd)
- self._loop._add_writer(self._sock_fd, self._write_ready)
- except Exception as exc:
- self._fatal_error(exc, 'Fatal read error on SSL transport')
- else:
- if data:
- self._protocol.data_received(data)
- else:
- try:
- if self._loop.get_debug():
- logger.debug("%r received EOF", self)
- keep_open = self._protocol.eof_received()
- if keep_open:
- logger.warning('returning true from eof_received() '
- 'has no effect when using ssl')
- finally:
- self.close()
-
- def _write_ready(self):
- if self._conn_lost:
- return
- if self._read_wants_write:
- self._read_wants_write = False
- self._read_ready()
-
- if not (self._paused or self._closing):
- self._loop._add_reader(self._sock_fd, self._read_ready)
-
- if self._buffer:
- try:
- n = self._sock.send(self._buffer)
- except (BlockingIOError, InterruptedError, ssl.SSLWantWriteError):
- n = 0
- except ssl.SSLWantReadError:
- n = 0
- self._loop._remove_writer(self._sock_fd)
- self._write_wants_read = True
- except Exception as exc:
- self._loop._remove_writer(self._sock_fd)
- self._buffer.clear()
- self._fatal_error(exc, 'Fatal write error on SSL transport')
- return
-
- if n:
- del self._buffer[:n]
-
- self._maybe_resume_protocol() # May append to buffer.
-
- if not self._buffer:
- self._loop._remove_writer(self._sock_fd)
- if self._closing:
- self._call_connection_lost(None)
-
- def write(self, data):
- if not isinstance(data, (bytes, bytearray, memoryview)):
- raise TypeError('data argument must be a bytes-like object, '
- 'not %r' % type(data).__name__)
- if not data:
- return
-
- if self._conn_lost:
- if self._conn_lost >= constants.LOG_THRESHOLD_FOR_CONNLOST_WRITES:
- logger.warning('socket.send() raised exception.')
- self._conn_lost += 1
- return
-
- if not self._buffer:
- self._loop._add_writer(self._sock_fd, self._write_ready)
-
- # Add it to the buffer.
- self._buffer.extend(data)
- self._maybe_pause_protocol()
-
- def can_write_eof(self):
- return False
-
-
class _SelectorDatagramTransport(_SelectorTransport):
_buffer_factory = collections.deque
# Client side may pass ssl=True to use a default
# context; in that case the sslcontext passed is None.
# The default is secure for client connections.
- if hasattr(ssl, 'create_default_context'):
- # Python 3.4+: use up-to-date strong settings.
- sslcontext = ssl.create_default_context()
- if not server_hostname:
- sslcontext.check_hostname = False
- else:
- # Fallback for Python 3.3.
- sslcontext = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
- sslcontext.options |= ssl.OP_NO_SSLv2
- sslcontext.options |= ssl.OP_NO_SSLv3
- sslcontext.set_default_verify_paths()
- sslcontext.verify_mode = ssl.CERT_REQUIRED
+ # Python 3.4+: use up-to-date strong settings.
+ sslcontext = ssl.create_default_context()
+ if not server_hostname:
+ sslcontext.check_hostname = False
return sslcontext
-def _is_sslproto_available():
- return hasattr(ssl, "MemoryBIO")
-
-
# States of an _SSLPipe.
_UNWRAPPED = "UNWRAPPED"
_DO_HANDSHAKE = "DO_HANDSHAKE"
sock.family = family
sock.gettimeout.return_value = 0.0
return sock
-
-
-def force_legacy_ssl_support():
- return mock.patch('asyncio.sslproto._is_sslproto_available',
- return_value=False)
self._test_create_ssl_connection(httpd, create_connection,
peername=httpd.address)
- def test_legacy_create_ssl_connection(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_ssl_connection()
-
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_ssl_unix_connection(self):
check_sockname,
peername=httpd.address)
- def test_legacy_create_ssl_unix_connection(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_ssl_unix_connection()
-
def test_create_connection_local_addr(self):
with test_utils.run_test_server() as httpd:
port = support.find_unused_port()
# stop serving
server.close()
- def test_legacy_create_server_ssl(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_server_ssl()
-
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_unix_server_ssl(self):
# stop serving
server.close()
- def test_legacy_create_unix_server_ssl(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_unix_server_ssl()
-
@unittest.skipIf(ssl is None, 'No ssl module')
def test_create_server_ssl_verify_failed(self):
proto = MyProto(loop=self.loop)
self.assertIsNone(proto.transport)
server.close()
- def test_legacy_create_server_ssl_verify_failed(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_server_ssl_verify_failed()
-
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_unix_server_ssl_verify_failed(self):
self.assertIsNone(proto.transport)
server.close()
-
- def test_legacy_create_unix_server_ssl_verify_failed(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_unix_server_ssl_verify_failed()
-
@unittest.skipIf(ssl is None, 'No ssl module')
def test_create_server_ssl_match_failed(self):
proto = MyProto(loop=self.loop)
proto.transport.close()
server.close()
- def test_legacy_create_server_ssl_match_failed(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_server_ssl_match_failed()
-
@unittest.skipIf(ssl is None, 'No ssl module')
@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')
def test_create_unix_server_ssl_verified(self):
server.close()
self.loop.run_until_complete(proto.done)
- def test_legacy_create_unix_server_ssl_verified(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_unix_server_ssl_verified()
-
@unittest.skipIf(ssl is None, 'No ssl module')
def test_create_server_ssl_verified(self):
proto = MyProto(loop=self.loop)
server.close()
self.loop.run_until_complete(proto.done)
- def test_legacy_create_server_ssl_verified(self):
- with test_utils.force_legacy_ssl_support():
- self.test_create_server_ssl_verified()
-
def test_create_server_sock(self):
proto = asyncio.Future(loop=self.loop)
def create_event_loop(self):
return asyncio.ProactorEventLoop()
- if not sslproto._is_sslproto_available():
- def test_create_ssl_connection(self):
- raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
- def test_create_server_ssl(self):
- raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
- def test_create_server_ssl_verify_failed(self):
- raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
- def test_create_server_ssl_match_failed(self):
- raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
- def test_create_server_ssl_verified(self):
- raise unittest.SkipTest("need python 3.5 (ssl.MemoryBIO)")
-
- def test_legacy_create_ssl_connection(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
- def test_legacy_create_server_ssl(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
- def test_legacy_create_server_ssl_verify_failed(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
- def test_legacy_create_server_ssl_match_failed(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
- def test_legacy_create_server_ssl_verified(self):
- raise unittest.SkipTest("IocpEventLoop incompatible with legacy SSL")
-
def test_reader_callback(self):
raise unittest.SkipTest("IocpEventLoop does not have add_reader()")
from asyncio import test_utils
from asyncio.selector_events import BaseSelectorEventLoop
from asyncio.selector_events import _SelectorTransport
-from asyncio.selector_events import _SelectorSslTransport
from asyncio.selector_events import _SelectorSocketTransport
from asyncio.selector_events import _SelectorDatagramTransport
remove_writer.assert_called_with(self.sock_fd)
-@unittest.skipIf(ssl is None, 'No ssl module')
-class SelectorSslTransportTests(test_utils.TestCase):
-
- def setUp(self):
- super().setUp()
- self.loop = self.new_test_loop()
- self.protocol = test_utils.make_test_protocol(asyncio.Protocol)
- self.sock = mock.Mock(socket.socket)
- self.sock.fileno.return_value = 7
- self.sslsock = mock.Mock()
- self.sslsock.fileno.return_value = 1
- self.sslcontext = mock.Mock()
- self.sslcontext.wrap_socket.return_value = self.sslsock
-
- def ssl_transport(self, waiter=None, server_hostname=None):
- transport = _SelectorSslTransport(self.loop, self.sock, self.protocol,
- self.sslcontext, waiter=waiter,
- server_hostname=server_hostname)
- self.addCleanup(close_transport, transport)
- return transport
-
- def _make_one(self, create_waiter=None):
- transport = self.ssl_transport()
- self.sock.reset_mock()
- self.sslsock.reset_mock()
- self.sslcontext.reset_mock()
- self.loop.reset_counters()
- return transport
-
- def test_on_handshake(self):
- waiter = asyncio.Future(loop=self.loop)
- tr = self.ssl_transport(waiter=waiter)
- self.assertTrue(self.sslsock.do_handshake.called)
- self.loop.assert_reader(1, tr._read_ready)
- test_utils.run_briefly(self.loop)
- self.assertIsNone(waiter.result())
-
- def test_on_handshake_reader_retry(self):
- self.loop.set_debug(False)
- self.sslsock.do_handshake.side_effect = ssl.SSLWantReadError
- transport = self.ssl_transport()
- self.loop.assert_reader(1, transport._on_handshake, None)
-
- def test_on_handshake_writer_retry(self):
- self.loop.set_debug(False)
- self.sslsock.do_handshake.side_effect = ssl.SSLWantWriteError
- transport = self.ssl_transport()
- self.loop.assert_writer(1, transport._on_handshake, None)
-
- def test_on_handshake_exc(self):
- exc = ValueError()
- self.sslsock.do_handshake.side_effect = exc
- with test_utils.disable_logger():
- waiter = asyncio.Future(loop=self.loop)
- self.ssl_transport(waiter=waiter)
- self.assertTrue(waiter.done())
- self.assertIs(exc, waiter.exception())
- self.assertTrue(self.sslsock.close.called)
-
- def test_on_handshake_base_exc(self):
- waiter = asyncio.Future(loop=self.loop)
- transport = self.ssl_transport(waiter=waiter)
- exc = BaseException()
- self.sslsock.do_handshake.side_effect = exc
- with test_utils.disable_logger():
- self.assertRaises(BaseException, transport._on_handshake, 0)
- self.assertTrue(self.sslsock.close.called)
- self.assertTrue(waiter.done())
- self.assertIs(exc, waiter.exception())
-
- def test_cancel_handshake(self):
- # Python issue #23197: cancelling a handshake must not raise an
- # exception or log an error, even if the handshake failed
- waiter = asyncio.Future(loop=self.loop)
- transport = self.ssl_transport(waiter=waiter)
- waiter.cancel()
- exc = ValueError()
- self.sslsock.do_handshake.side_effect = exc
- with test_utils.disable_logger():
- transport._on_handshake(0)
- transport.close()
- test_utils.run_briefly(self.loop)
-
- def test_pause_resume_reading(self):
- tr = self._make_one()
- self.assertFalse(tr._paused)
- self.loop.assert_reader(1, tr._read_ready)
- tr.pause_reading()
- self.assertTrue(tr._paused)
- self.assertFalse(1 in self.loop.readers)
- tr.resume_reading()
- self.assertFalse(tr._paused)
- self.loop.assert_reader(1, tr._read_ready)
- with self.assertRaises(RuntimeError):
- tr.resume_reading()
-
- def test_write(self):
- transport = self._make_one()
- transport.write(b'data')
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_bytearray(self):
- transport = self._make_one()
- data = bytearray(b'data')
- transport.write(data)
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
- self.assertEqual(data, bytearray(b'data')) # Hasn't been mutated.
- self.assertIsNot(data, transport._buffer) # Hasn't been incorporated.
-
- def test_write_memoryview(self):
- transport = self._make_one()
- data = memoryview(b'data')
- transport.write(data)
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_no_data(self):
- transport = self._make_one()
- transport._buffer.extend(b'data')
- transport.write(b'')
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_str(self):
- transport = self._make_one()
- self.assertRaises(TypeError, transport.write, 'str')
-
- def test_write_closing(self):
- transport = self._make_one()
- transport.close()
- self.assertEqual(transport._conn_lost, 1)
- transport.write(b'data')
- self.assertEqual(transport._conn_lost, 2)
-
- @mock.patch('asyncio.selector_events.logger')
- def test_write_exception(self, m_log):
- transport = self._make_one()
- transport._conn_lost = 1
- transport.write(b'data')
- self.assertEqual(transport._buffer, list_to_buffer())
- transport.write(b'data')
- transport.write(b'data')
- transport.write(b'data')
- transport.write(b'data')
- m_log.warning.assert_called_with('socket.send() raised exception.')
-
- def test_read_ready_recv(self):
- self.sslsock.recv.return_value = b'data'
- transport = self._make_one()
- transport._read_ready()
- self.assertTrue(self.sslsock.recv.called)
- self.assertEqual((b'data',), self.protocol.data_received.call_args[0])
-
- def test_read_ready_write_wants_read(self):
- self.loop._add_writer = mock.Mock()
- self.sslsock.recv.side_effect = BlockingIOError
- transport = self._make_one()
- transport._write_wants_read = True
- transport._write_ready = mock.Mock()
- transport._buffer.extend(b'data')
- transport._read_ready()
-
- self.assertFalse(transport._write_wants_read)
- transport._write_ready.assert_called_with()
- self.loop._add_writer.assert_called_with(
- transport._sock_fd, transport._write_ready)
-
- def test_read_ready_recv_eof(self):
- self.sslsock.recv.return_value = b''
- transport = self._make_one()
- transport.close = mock.Mock()
- transport._read_ready()
- transport.close.assert_called_with()
- self.protocol.eof_received.assert_called_with()
-
- def test_read_ready_recv_conn_reset(self):
- err = self.sslsock.recv.side_effect = ConnectionResetError()
- transport = self._make_one()
- transport._force_close = mock.Mock()
- with test_utils.disable_logger():
- transport._read_ready()
- transport._force_close.assert_called_with(err)
-
- def test_read_ready_recv_retry(self):
- self.sslsock.recv.side_effect = ssl.SSLWantReadError
- transport = self._make_one()
- transport._read_ready()
- self.assertTrue(self.sslsock.recv.called)
- self.assertFalse(self.protocol.data_received.called)
-
- self.sslsock.recv.side_effect = BlockingIOError
- transport._read_ready()
- self.assertFalse(self.protocol.data_received.called)
-
- self.sslsock.recv.side_effect = InterruptedError
- transport._read_ready()
- self.assertFalse(self.protocol.data_received.called)
-
- def test_read_ready_recv_write(self):
- self.loop._remove_reader = mock.Mock()
- self.loop._add_writer = mock.Mock()
- self.sslsock.recv.side_effect = ssl.SSLWantWriteError
- transport = self._make_one()
- transport._read_ready()
- self.assertFalse(self.protocol.data_received.called)
- self.assertTrue(transport._read_wants_write)
-
- self.loop._remove_reader.assert_called_with(transport._sock_fd)
- self.loop._add_writer.assert_called_with(
- transport._sock_fd, transport._write_ready)
-
- def test_read_ready_recv_exc(self):
- err = self.sslsock.recv.side_effect = OSError()
- transport = self._make_one()
- transport._fatal_error = mock.Mock()
- transport._read_ready()
- transport._fatal_error.assert_called_with(
- err,
- 'Fatal read error on SSL transport')
-
- def test_write_ready_send(self):
- self.sslsock.send.return_value = 4
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
- transport._write_ready()
- self.assertEqual(list_to_buffer(), transport._buffer)
- self.assertTrue(self.sslsock.send.called)
-
- def test_write_ready_send_none(self):
- self.sslsock.send.return_value = 0
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data1', b'data2'])
- transport._write_ready()
- self.assertTrue(self.sslsock.send.called)
- self.assertEqual(list_to_buffer([b'data1data2']), transport._buffer)
-
- def test_write_ready_send_partial(self):
- self.sslsock.send.return_value = 2
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data1', b'data2'])
- transport._write_ready()
- self.assertTrue(self.sslsock.send.called)
- self.assertEqual(list_to_buffer([b'ta1data2']), transport._buffer)
-
- def test_write_ready_send_closing_partial(self):
- self.sslsock.send.return_value = 2
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data1', b'data2'])
- transport._write_ready()
- self.assertTrue(self.sslsock.send.called)
- self.assertFalse(self.sslsock.close.called)
-
- def test_write_ready_send_closing(self):
- self.sslsock.send.return_value = 4
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
- transport.close()
- transport._write_ready()
- self.protocol.connection_lost.assert_called_with(None)
-
- def test_write_ready_send_closing_empty_buffer(self):
- self.sslsock.send.return_value = 4
- call_soon = self.loop.call_soon = mock.Mock()
- transport = self._make_one()
- transport._buffer = list_to_buffer()
- transport.close()
- transport._write_ready()
- call_soon.assert_called_with(transport._call_connection_lost, None)
-
- def test_write_ready_send_retry(self):
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
-
- self.sslsock.send.side_effect = ssl.SSLWantWriteError
- transport._write_ready()
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- self.sslsock.send.side_effect = BlockingIOError()
- transport._write_ready()
- self.assertEqual(list_to_buffer([b'data']), transport._buffer)
-
- def test_write_ready_send_read(self):
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
-
- self.loop._remove_writer = mock.Mock()
- self.sslsock.send.side_effect = ssl.SSLWantReadError
- transport._write_ready()
- self.assertFalse(self.protocol.data_received.called)
- self.assertTrue(transport._write_wants_read)
- self.loop._remove_writer.assert_called_with(transport._sock_fd)
-
- def test_write_ready_send_exc(self):
- err = self.sslsock.send.side_effect = OSError()
-
- transport = self._make_one()
- transport._buffer = list_to_buffer([b'data'])
- transport._fatal_error = mock.Mock()
- transport._write_ready()
- transport._fatal_error.assert_called_with(
- err,
- 'Fatal write error on SSL transport')
- self.assertEqual(list_to_buffer(), transport._buffer)
-
- def test_write_ready_read_wants_write(self):
- self.loop._add_reader = mock.Mock()
- self.sslsock.send.side_effect = BlockingIOError
- transport = self._make_one()
- transport._read_wants_write = True
- transport._read_ready = mock.Mock()
- transport._write_ready()
-
- self.assertFalse(transport._read_wants_write)
- transport._read_ready.assert_called_with()
- self.loop._add_reader.assert_called_with(
- transport._sock_fd, transport._read_ready)
-
- def test_write_eof(self):
- tr = self._make_one()
- self.assertFalse(tr.can_write_eof())
- self.assertRaises(NotImplementedError, tr.write_eof)
-
- def check_close(self):
- tr = self._make_one()
- tr.close()
-
- self.assertTrue(tr.is_closing())
- self.assertEqual(1, self.loop.remove_reader_count[1])
- self.assertEqual(tr._conn_lost, 1)
-
- tr.close()
- self.assertEqual(tr._conn_lost, 1)
- self.assertEqual(1, self.loop.remove_reader_count[1])
-
- test_utils.run_briefly(self.loop)
-
- def test_close(self):
- self.check_close()
- self.assertTrue(self.protocol.connection_made.called)
- self.assertTrue(self.protocol.connection_lost.called)
-
- def test_close_not_connected(self):
- self.sslsock.do_handshake.side_effect = ssl.SSLWantReadError
- self.check_close()
- self.assertFalse(self.protocol.connection_made.called)
- self.assertFalse(self.protocol.connection_lost.called)
-
- @unittest.skipIf(ssl is None, 'No SSL support')
- def test_server_hostname(self):
- self.ssl_transport(server_hostname='localhost')
- self.sslcontext.wrap_socket.assert_called_with(
- self.sock, do_handshake_on_connect=False, server_side=False,
- server_hostname='localhost')
-
-
-class SelectorSslWithoutSslTransportTests(unittest.TestCase):
-
- @mock.patch('asyncio.selector_events.ssl', None)
- def test_ssl_transport_requires_ssl_module(self):
- Mock = mock.Mock
- with self.assertRaises(RuntimeError):
- _SelectorSslTransport(Mock(), Mock(), Mock(), Mock())
-
-
class SelectorDatagramTransportTests(test_utils.TestCase):
def setUp(self):
--- /dev/null
+Drop legacy SSL transport from asyncio, ssl.MemoryBIO is always used anyway.
projects / python / commitdiff