[3.12] gh-118950: Fix SSLProtocol.connection_lost not being called when OSError is thrown (GH-118960) (#125932)

gh-118950: Fix SSLProtocol.connection_lost not being called when OSError is thrown (GH-118960)

(cherry picked from commit 3f24bde0b6)

Co-authored-by: Javad Shafique <javadshafique@hotmail.com>
Co-authored-by: Kumar Aditya <kumaraditya@python.org>

Lib/test/test_asyncio/test_sslproto.py

@@ -109,6 +109,54 @@ class SslProtoHandshakeTests(test_utils.TestCase):
         test_utils.run_briefly(self.loop)
         self.assertIsInstance(waiter.exception(), ConnectionAbortedError)
 
+    def test_connection_lost_when_busy(self):
+        # gh-118950: SSLProtocol.connection_lost not being called when OSError
+        # is thrown on asyncio.write.
+        sock = mock.Mock()
+        sock.fileno = mock.Mock(return_value=12345)
+        sock.send = mock.Mock(side_effect=BrokenPipeError)
+
+        # construct StreamWriter chain that contains loop dependant logic this emulates
+        # what _make_ssl_transport() does in BaseSelectorEventLoop
+        reader = asyncio.StreamReader(limit=2 ** 16, loop=self.loop)
+        protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop)
+        ssl_proto = self.ssl_protocol(proto=protocol)
+
+        # emulate reading decompressed data
+        sslobj = mock.Mock()
+        sslobj.read.side_effect = ssl.SSLWantReadError
+        sslobj.write.side_effect = ssl.SSLWantReadError
+        ssl_proto._sslobj = sslobj
+
+        # emulate outgoing data
+        data = b'An interesting message'
+
+        outgoing = mock.Mock()
+        outgoing.read = mock.Mock(return_value=data)
+        outgoing.pending = len(data)
+        ssl_proto._outgoing = outgoing
+
+        # use correct socket transport to initialize the SSLProtocol
+        self.loop._make_socket_transport(sock, ssl_proto)
+
+        transport = ssl_proto._app_transport
+        writer = asyncio.StreamWriter(transport, protocol, reader, self.loop)
+
+        async def main():
+            # writes data to transport
+            async def write():
+                writer.write(data)
+                await writer.drain()
+
+            # try to write for the first time
+            await write()
+            # try to write for the second time, this raises as the connection_lost
+            # callback should be done with error
+            with self.assertRaises(ConnectionResetError):
+                await write()
+
+        self.loop.run_until_complete(main())
+
     def test_close_during_handshake(self):
         # bpo-29743 Closing transport during handshake process leaks socket
         waiter = self.loop.create_future()