gh-159875: Add tests for an SSL transport closed after connection loss

An application that closes its transport from connection_lost(), as aiohttp
does, must be left with a transport whose write(), get_write_buffer_size()
and set_protocol() behave as after closing a live connection, and
loop.sendfile()'s fallback must then raise ConnectionError rather than
AttributeError from set_protocol().

The sendfile test reads the file synchronously, so that the fallback is
waiting in drain() when the connection is lost; a read in the executor that
completes in the same loop iteration as the reset lets the fallback finish
first and hides the bug.

test_create_connection_memory_leak now keeps the transport past its check,
as its comment says it covers, so that it would see an SSLContext kept alive
by a transport the application still holds.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: cornopaez

Lib/test/test_asyncio/test_sendfile.py

@@ -553,6 +553,38 @@ def test_sendfile_ssl_close_peer_after_receiving(self):
         self.assertEqual(srv_proto.data, self.DATA)
         self.assertEqual(self.file.tell(), len(self.DATA))
 
+    def test_sendfile_ssl_close_peer_in_the_middle_closes_transport(self):
+        # The protocol closes its transport when the connection is lost, as
+        # aiohttp's does.  That close() must leave a transport on which the
+        # fallback can still restore the protocol: sendfile() raises
+        # ConnectionError, not AttributeError.
+        srv_proto, cli_proto = self.prepare_sendfile(
+            is_ssl=True, close_after=1024)
+        connection_lost = cli_proto.connection_lost
+
+        def close_on_lost(exc):
+            cli_proto.transport.close()
+            connection_lost(exc)
+
+        cli_proto.connection_lost = close_on_lost
+        # Read the file synchronously, so that the fallback only waits in
+        # drain().  A read in the executor that completes in the same loop
+        # iteration as the reset lets the fallback finish before close() is
+        # called, which hides the bug.
+        def run_in_executor(executor, func, *args):
+            fut = self.loop.create_future()
+            fut.set_result(func(*args))
+            return fut
+
+        with mock.patch.object(self.loop, 'run_in_executor', run_in_executor):
+            with self.assertRaises(ConnectionError):
+                self.run_loop(
+                    self.loop.sendfile(cli_proto.transport, self.file))
+        self.run_loop(srv_proto.done)
+        self.run_loop(cli_proto.done)
+        self.assertTrue(1024 <= srv_proto.nbytes < len(self.DATA),
+                        srv_proto.nbytes)
+
     # On Solaris, lowering SO_RCVBUF on a TCP connection after it has been
     # established has no effect. Due to its age, this bug affects both Oracle
     # Solaris as well as all other OpenSolaris forks (unless they fixed it

Lib/test/test_asyncio/test_ssl.py

@@ -788,16 +788,21 @@ async def client(addr):
             await on_eof
 
             tr.close()
+            return tr
 
         with self.tcp_server(serve, timeout=self.TIMEOUT) as srv:
-            self.loop.run_until_complete(
+            tr = self.loop.run_until_complete(
                 asyncio.wait_for(client(srv.addr),
                                  timeout=support.SHORT_TIMEOUT))
+        # Let the protocol's connection_lost() run.
+        self.loop.run_until_complete(asyncio.sleep(0))
 
         # No garbage is left for SSL client from loop.create_connection, even
         # if user stores the SSLTransport in corresponding protocol instance
+        # (and keeps the transport after the connection is lost).
         client_context = weakref.ref(client_context)
         self.assertIsNone(client_context())
+        del tr
 
     def test_start_tls_client_buf_proto_1(self):
         HELLO_MSG = b'1' * self.PAYLOAD_SIZE

Lib/test/test_asyncio/test_sslproto.py

@@ -265,6 +265,26 @@ def test_write_after_closing(self):
         # should not raise
         self.assertIsNone(transp.write(b'data'))
 
+    def test_transport_usable_after_connection_lost_and_close(self):
+        # connection_lost() marks the app transport closed, so the
+        # application's own close() afterwards (often made from its
+        # protocol's connection_lost()) must not be taken as a second
+        # close() that leaves the transport without its SSL protocol.
+        ssl_proto = self.ssl_protocol()
+        self.connection_made(ssl_proto)
+        transp = ssl_proto._app_transport
+        ssl_proto.connection_lost(None)
+        transp.close()
+
+        # should not raise, as after closing a live connection
+        self.assertIsNone(transp.write(b'data'))
+        self.assertEqual(transp.get_write_buffer_size(), 0)
+        new_proto = asyncio.Protocol()
+        transp.set_protocol(new_proto)  # as loop.sendfile()'s fallback does
+        self.assertIs(transp.get_protocol(), new_proto)
+        transp.abort()
+        self.assertTrue(transp.is_closing())
+
 
 ##############################################################################
 # Start TLS Tests