gh-155027: Make test_asyncio's socket harness able to fail a test

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:

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
 
 
 ##############################################################################

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.

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.