diff --git a/Lib/asyncio/sslproto.py b/Lib/asyncio/sslproto.py index 84e82fc69cc5fdc..a44518dce66deb6 100644 --- a/Lib/asyncio/sslproto.py +++ b/Lib/asyncio/sslproto.py @@ -323,6 +323,8 @@ def __init__(self, loop, app_protocol, sslcontext, waiter, self._outgoing = ssl.MemoryBIO() self._state = SSLProtocolState.UNWRAPPED self._conn_lost = 0 # Set when connection_lost called + self._shutdown_exc = None + self._shutdown_close_pending = False if call_connection_made: self._app_state = AppProtocolState.STATE_INIT else: @@ -397,6 +399,11 @@ def connection_lost(self, exc): meaning a regular EOF is received or the connection was aborted or closed). """ + if exc is None and self._shutdown_exc is not None: + exc = self._shutdown_exc + self._shutdown_exc = None + self._shutdown_close_pending = False + self._write_backlog.clear() self._outgoing.read() self._conn_lost += 1 @@ -669,14 +676,21 @@ def _do_shutdown(self): self._on_shutdown_complete(None) def _on_shutdown_complete(self, shutdown_exc): - if self._shutdown_timeout_handle is not None: - self._shutdown_timeout_handle.cancel() - self._shutdown_timeout_handle = None + # close() lets the raw transport flush data queued by + # _process_outgoing(). _fatal_error() would force-close it and + # discard the close_notify that shutdown just produced. Keep the + # shutdown timeout active until connection_lost() bounds the drain. + if shutdown_exc is not None: + self._shutdown_exc = shutdown_exc + self._shutdown_close_pending = True + if not self._ssl_writing_paused: + self._loop.call_soon(self._close_transport) - if shutdown_exc: - self._fatal_error(shutdown_exc) - else: - self._loop.call_soon(self._transport.close) + def _close_transport(self): + if self._shutdown_close_pending: + self._shutdown_close_pending = False + if self._transport is not None: + self._transport.close() def _abort(self, exc): self._set_state(SSLProtocolState.UNWRAPPED) @@ -927,6 +941,8 @@ def resume_writing(self): assert self._ssl_writing_paused self._ssl_writing_paused = False self._process_outgoing() + if self._shutdown_close_pending: + self._loop.call_soon(self._close_transport) def _fatal_error(self, exc, message='Fatal error on transport'): if self._transport: diff --git a/Lib/test/test_asyncio/functional.py b/Lib/test/test_asyncio/functional.py index 96dc9ab44010670..47a3ab8c306edd9 100644 --- a/Lib/test/test_asyncio/functional.py +++ b/Lib/test/test_asyncio/functional.py @@ -28,6 +28,7 @@ def setUp(self): self.loop.set_exception_handler(self.loop_exception_handler) self.__unhandled_exceptions = [] + self.__abort_exception = None def tearDown(self): try: @@ -38,9 +39,13 @@ def tearDown(self): pprint.pprint(self.__unhandled_exceptions) self.fail('unexpected calls to loop.call_exception_handler()') + if self.__abort_exception is not None: + raise self.__abort_exception + finally: asyncio.set_event_loop(None) self.loop = None + self.__abort_exception = None def tcp_server(self, server_prog, *, family=socket.AF_INET, @@ -104,10 +109,18 @@ def unix_sock_name(self): pass def _abort_socket_test(self, ex): + # This runs in the client/server thread, not the main thread, so + # it must not call self.fail(): the AssertionError would escape + # Thread.run() without failing the test. Stash the exception and + # let tearDown() re-raise it on the main thread. try: - self.loop.stop() + self.loop.call_soon_threadsafe(self.loop.stop) + except RuntimeError: + # The loop is already closed; nothing left to stop. + pass finally: - self.fail(ex) + if self.__abort_exception is None: + self.__abort_exception = ex ############################################################################## diff --git a/Lib/test/test_asyncio/test_sslproto.py b/Lib/test/test_asyncio/test_sslproto.py index 656cdf570fad7b1..4c6feeb50511b8b 100644 --- a/Lib/test/test_asyncio/test_sslproto.py +++ b/Lib/test/test_asyncio/test_sslproto.py @@ -96,6 +96,77 @@ def test_fatal_error_no_name_error(self): # Restore error logging. log.logger.setLevel(log_level) + def test_shutdown_error_closes_after_flushing(self): + app_proto = mock.Mock(spec=asyncio.Protocol) + app_proto.eof_received.return_value = False + ssl_proto = self.ssl_protocol(proto=app_proto) + ssl_proto._state = sslproto.SSLProtocolState.SHUTDOWN + ssl_proto._app_state = sslproto.AppProtocolState.STATE_CON_MADE + transport = mock.Mock() + ssl_proto._transport = transport + ssl_proto._sslobj = mock.Mock() + shutdown_exc = ssl.SSLError(ssl.SSL_ERROR_SSL, 'shutdown failed') + ssl_proto._sslobj.unwrap.side_effect = shutdown_exc + ssl_proto._outgoing = mock.Mock() + ssl_proto._outgoing.read.side_effect = [b'close notify', b'', b''] + ssl_proto._outgoing.pending = 0 + timeout_handle = mock.Mock() + ssl_proto._shutdown_timeout_handle = timeout_handle + + ssl_proto._do_shutdown() + ssl_proto.eof_received() + + transport.write.assert_called_once_with(b'close notify') + transport._force_close.assert_not_called() + test_utils.run_briefly(self.loop) + transport.close.assert_called_once_with() + timeout_handle.cancel.assert_not_called() + + ssl_proto.connection_lost(None) + test_utils.run_briefly(self.loop) + app_proto.connection_lost.assert_called_once_with(shutdown_exc) + timeout_handle.cancel.assert_called_once_with() + + def test_shutdown_waits_for_resume_writing_before_close(self): + app_proto = mock.Mock(spec=asyncio.Protocol) + ssl_proto = self.ssl_protocol(proto=app_proto) + ssl_proto._state = sslproto.SSLProtocolState.SHUTDOWN + ssl_proto._app_state = sslproto.AppProtocolState.STATE_CON_MADE + transport = mock.Mock() + ssl_proto._transport = transport + ssl_proto._sslobj = mock.Mock() + shutdown_exc = ssl.SSLError(ssl.SSL_ERROR_SSL, 'shutdown failed') + ssl_proto._sslobj.unwrap.side_effect = shutdown_exc + ssl_proto._outgoing = mock.Mock() + ssl_proto._outgoing.read.return_value = b'close notify' + ssl_proto._outgoing.pending = 0 + ssl_proto._ssl_writing_paused = True + + ssl_proto._do_shutdown() + test_utils.run_briefly(self.loop) + + transport.write.assert_not_called() + transport.close.assert_not_called() + + ssl_proto.resume_writing() + transport.write.assert_called_once_with(b'close notify') + test_utils.run_briefly(self.loop) + transport.close.assert_called_once_with() + + def test_shutdown_raw_error_takes_precedence(self): + app_proto = mock.Mock(spec=asyncio.Protocol) + ssl_proto = self.ssl_protocol(proto=app_proto) + ssl_proto._state = sslproto.SSLProtocolState.SHUTDOWN + ssl_proto._app_state = sslproto.AppProtocolState.STATE_CON_MADE + ssl_proto._shutdown_exc = ssl.SSLError( + ssl.SSL_ERROR_SSL, 'shutdown failed') + raw_exc = ConnectionResetError('raw write failed') + + ssl_proto.connection_lost(raw_exc) + test_utils.run_briefly(self.loop) + + app_proto.connection_lost.assert_called_once_with(raw_exc) + def test_connection_lost(self): # From issue #472. # yield from waiter hang if lost_connection was called. diff --git a/Misc/NEWS.d/next/Tests/2026-08-01-10-30-00.gh-issue-155027.Kq7Wm2.rst b/Misc/NEWS.d/next/Tests/2026-08-01-10-30-00.gh-issue-155027.Kq7Wm2.rst new file mode 100644 index 000000000000000..f23b541d9e3c2bc --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2026-08-01-10-30-00.gh-issue-155027.Kq7Wm2.rst @@ -0,0 +1,7 @@ +Fix ``test_asyncio``'s socket test harness so that a failure in the client +or server thread actually fails the test. ``_abort_socket_test()`` called +``self.fail()`` from a worker thread, where the resulting +:exc:`AssertionError` cannot fail the test; it now records the exception and +re-raises it on the main thread. It also stops the event loop with +:meth:`~asyncio.loop.call_soon_threadsafe` rather than calling +:meth:`~asyncio.loop.stop` directly from a non-main thread.