[Core] Support timeout error in requests transport

sdk/core/azure-core/azure/core/pipeline/_tools.py

@@ -80,7 +80,5 @@ def handle_non_stream_rest_response(response: HttpResponse) -> None:
     """
     try:
         response.read()
+    finally:
         response.close()
-    except Exception as exc:
-        response.close()
-        raise exc

sdk/core/azure-core/azure/core/pipeline/_tools_async.py

@@ -67,7 +67,5 @@ async def handle_no_stream_rest_response(response: "RestAsyncHttpResponse") -> N
     """
     try:
         await response.read()
+    finally:
         await response.close()
-    except Exception as exc:
-        await response.close()
-        raise exc

sdk/core/azure-core/azure/core/pipeline/transport/_requests_basic.py

@@ -46,7 +46,9 @@
 from azure.core.configuration import ConnectionConfiguration
 from azure.core.exceptions import (
     ServiceRequestError,
+    ServiceRequestTimeoutError,
     ServiceResponseError,
+    ServiceResponseTimeoutError,
     IncompleteReadError,
     HttpResponseError,
     DecodeError,
@@ -85,7 +87,7 @@ def _read_raw_stream(response, chunk_size=1):
         except CoreDecodeError as e:
             raise DecodeError(e, error=e) from e
         except ReadTimeoutError as e:
-            raise ServiceRequestError(e, error=e) from e
+            raise ServiceResponseTimeoutError(e, error=e) from e
     else:
         # Standard file-like object.
         while True:
@@ -202,6 +204,14 @@ def __next__(self):
             _LOGGER.warning("Unable to stream download.")
             internal_response.close()
             raise HttpResponseError(err, error=err) from err
+        except requests.ConnectionError as err:
+            internal_response.close()
+            if err.args and isinstance(err.args[0], ReadTimeoutError):
+                raise ServiceResponseTimeoutError(err, error=err) from err
+            raise ServiceResponseError(err, error=err) from err
+        except requests.RequestException as err:
+            internal_response.close()
+            raise ServiceResponseError(err, error=err) from err
         except Exception as err:
             _LOGGER.warning("Unable to stream download.")
             internal_response.close()
@@ -384,13 +394,14 @@ def send(  # pylint: disable=too-many-statements
                     "Please report this issue to https://github.com/Azure/azure-sdk-for-python/issues."
                 ) from err
             raise
-        except (
-            NewConnectionError,
-            ConnectTimeoutError,
-        ) as err:
+        except NewConnectionError as err:
             error = ServiceRequestError(err, error=err)
+        except ConnectTimeoutError as err:
+            error = ServiceRequestTimeoutError(err, error=err)
+        except requests.exceptions.ConnectTimeout as err:
+            error = ServiceRequestTimeoutError(err, error=err)
         except requests.exceptions.ReadTimeout as err:
-            error = ServiceResponseError(err, error=err)
+            error = ServiceResponseTimeoutError(err, error=err)
         except requests.exceptions.ConnectionError as err:
             if err.args and isinstance(err.args[0], ProtocolError):
                 error = ServiceResponseError(err, error=err)
@@ -405,7 +416,7 @@ def send(  # pylint: disable=too-many-statements
                 _LOGGER.warning("Unable to stream download.")
                 error = HttpResponseError(err, error=err)
         except requests.RequestException as err:
-            error = ServiceRequestError(err, error=err)
+            error = ServiceResponseError(err, error=err)
 
         if error:
             raise error

sdk/core/azure-core/tests/test_basic_transport.py

@@ -6,19 +6,31 @@
 from http.client import HTTPConnection
 from collections import OrderedDict
 import sys
+import logging
+import pytest
 
 try:
     from unittest import mock
 except ImportError:
     import mock
 
+from urllib3.util import connection
+from urllib3.response import HTTPResponse
+from urllib3.connection import HTTPConnection
+from socket import timeout as SocketTimeout
+
 from azure.core.pipeline.transport import HttpResponse as PipelineTransportHttpResponse, RequestsTransport
 from azure.core.pipeline.transport._base import HttpTransport, _deserialize_response, _urljoin
 from azure.core.pipeline.policies import HeadersPolicy
 from azure.core.pipeline import Pipeline
-from azure.core.exceptions import HttpResponseError
-import logging
-import pytest
+from azure.core.exceptions import (
+    HttpResponseError,
+    ServiceRequestError,
+    ServiceResponseError,
+    ServiceRequestTimeoutError,
+    ServiceResponseTimeoutError,
+)
+
 from utils import (
     HTTP_REQUESTS,
     request_and_responses_product,
@@ -1322,3 +1334,46 @@ def test_close_too_soon_works_fine(caplog, port, http_request):
     result = transport.send(request)
 
     assert result  # No exception is good enough here
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_requests_timeout_response(caplog, port, http_request):
+    transport = RequestsTransport()
+
+    request = http_request("GET", f"http://localhost:{port}/basic/string")
+
+    with mock.patch.object(HTTPConnection, "getresponse", side_effect=SocketTimeout) as mock_method:
+        with pytest.raises(ServiceResponseTimeoutError) as err:
+            transport.send(request, read_timeout=0.0001)
+
+        with pytest.raises(ServiceResponseError) as err:
+            transport.send(request, read_timeout=0.0001)
+
+        stream_request = http_request("GET", f"http://localhost:{port}/streams/basic")
+        with pytest.raises(ServiceResponseTimeoutError) as err:
+            transport.send(stream_request, stream=True, read_timeout=0.0001)
+
+    stream_resp = transport.send(stream_request, stream=True)
+    with mock.patch.object(HTTPResponse, "_handle_chunk", side_effect=SocketTimeout) as mock_method:
+        with pytest.raises(ServiceResponseTimeoutError) as err:
+            try:
+                stream_resp.read()
+            except AttributeError:
+                b"".join(stream_resp.stream_download(None))
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_requests_timeout_request(caplog, port, http_request):
+    transport = RequestsTransport()
+
+    request = http_request("GET", f"http://localhost:{port}/basic/string")
+
+    with mock.patch.object(connection, 'create_connection', side_effect=SocketTimeout) as mock_method:
+        with pytest.raises(ServiceRequestTimeoutError) as err:
+            transport.send(request, connection_timeout=0.0001)
+
+        with pytest.raises(ServiceRequestTimeoutError) as err:
+            transport.send(request, connection_timeout=0.0001)
+
+        stream_request = http_request("GET", f"http://localhost:{port}/streams/basic")
+        with pytest.raises(ServiceRequestTimeoutError) as err:
+            transport.send(stream_request, stream=True, connection_timeout=0.0001)