Add test coverage for proxy TLS handshake failure

Author: baizhufbb

tests/_async/test_http_proxy.py

@@ -276,3 +276,45 @@ def test_proxy_headers():
     assert proxy.headers == [
         (b"Proxy-Authorization", b"Basic dXNlcm5hbWU6cGFzc3dvcmQ=")
     ]
+
+
+@pytest.mark.anyio
+async def test_proxy_tunneling_tls_error():
+    """
+    Send an HTTPS request via a proxy, but the TLS handshake fails.
+    """
+    class BrokenTLSStream(AsyncMockStream):
+        async def start_tls(
+            self,
+            ssl_context: ssl.SSLContext,
+            server_hostname: typing.Optional[str] = None,
+            timeout: typing.Optional[float] = None,
+        ) -> AsyncNetworkStream:
+            raise OSError("TLS Failure")
+
+    class BrokenTLSBackend(AsyncMockBackend):
+        async def connect_tcp(
+            self,
+            host: str,
+            port: int,
+            timeout: typing.Optional[float] = None,
+            local_address: typing.Optional[str] = None,
+            socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None,
+        ) -> AsyncNetworkStream:
+            return BrokenTLSStream(list(self._buffer))
+
+    network_backend = BrokenTLSBackend(
+        [
+            b"HTTP/1.1 200 OK\r\n\r\n",
+        ]
+    )
+
+    async with AsyncConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        with pytest.raises(OSError, match="TLS Failure"):
+            await proxy.request("GET", "https://example.com/")
+        
+        assert not proxy.connections
+

tests/_sync/test_http_proxy.py

@@ -276,3 +276,45 @@ def test_proxy_headers():
     assert proxy.headers == [
         (b"Proxy-Authorization", b"Basic dXNlcm5hbWU6cGFzc3dvcmQ=")
     ]
+
+
+
+def test_proxy_tunneling_tls_error():
+    """
+    Send an HTTPS request via a proxy, but the TLS handshake fails.
+    """
+    class BrokenTLSStream(MockStream):
+        def start_tls(
+            self,
+            ssl_context: ssl.SSLContext,
+            server_hostname: typing.Optional[str] = None,
+            timeout: typing.Optional[float] = None,
+        ) -> NetworkStream:
+            raise OSError("TLS Failure")
+
+    class BrokenTLSBackend(MockBackend):
+        def connect_tcp(
+            self,
+            host: str,
+            port: int,
+            timeout: typing.Optional[float] = None,
+            local_address: typing.Optional[str] = None,
+            socket_options: typing.Optional[typing.Iterable[SOCKET_OPTION]] = None,
+        ) -> NetworkStream:
+            return BrokenTLSStream(list(self._buffer))
+
+    network_backend = BrokenTLSBackend(
+        [
+            b"HTTP/1.1 200 OK\r\n\r\n",
+        ]
+    )
+
+    with ConnectionPool(
+        proxy=Proxy("http://localhost:8080/"),
+        network_backend=network_backend,
+    ) as proxy:
+        with pytest.raises(OSError, match="TLS Failure"):
+            proxy.request("GET", "https://example.com/")
+        
+        assert not proxy.connections
+