More explicit error message if transport is already closed (#35559)

* Clarify error message if transport is already closed

* Add tests

* Add async impl and tests

* Slight clarification on async path

* If close before opening, don't fail

* Simplify the code a bit

* Pylint

* ChangeLog

* Thread safety

* Adapt to new black

sdk/core/azure-core/CHANGELOG.md

@@ -10,6 +10,8 @@
 
 ### Bugs Fixed
 
+- Raise correct exception if transport is used while already closed  #35559
+
 ### Other Changes
 
 - HTTP tracing spans will now include an `error.type` attribute if an error status code is returned.  #34619

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

@@ -112,6 +112,8 @@ def __init__(
             raise ValueError("session_owner cannot be False if no session is provided")
         self.connection_config = ConnectionConfiguration(**kwargs)
         self._use_env_settings = kwargs.pop("use_env_settings", True)
+        # See https://github.com/Azure/azure-sdk-for-python/issues/25640 to understand why we track this
+        self._has_been_opened = False
 
     async def __aenter__(self):
         await self.open()
@@ -126,26 +128,33 @@ async def __aexit__(
         await self.close()
 
     async def open(self):
-        """Opens the connection."""
-        if not self.session and self._session_owner:
-            jar = aiohttp.DummyCookieJar()
-            clientsession_kwargs = {
-                "trust_env": self._use_env_settings,
-                "cookie_jar": jar,
-                "auto_decompress": False,
-            }
-            if self._loop is not None:
-                clientsession_kwargs["loop"] = self._loop
-            self.session = aiohttp.ClientSession(**clientsession_kwargs)
-        # pyright has trouble to understand that self.session is not None, since we raised at worst in the init
-        self.session = cast(aiohttp.ClientSession, self.session)
+        if self._has_been_opened and not self.session:
+            raise ValueError(
+                "HTTP transport has already been closed. "
+                "You may check if you're calling a function outside of the `async with` of your client creation, "
+                "or if you called `await close()` on your client already."
+            )
+        if not self.session:
+            if self._session_owner:
+                jar = aiohttp.DummyCookieJar()
+                clientsession_kwargs = {
+                    "trust_env": self._use_env_settings,
+                    "cookie_jar": jar,
+                    "auto_decompress": False,
+                }
+                if self._loop is not None:
+                    clientsession_kwargs["loop"] = self._loop
+                self.session = aiohttp.ClientSession(**clientsession_kwargs)
+            else:
+                raise ValueError("session_owner cannot be False and no session is available")
+
+        self._has_been_opened = True
         await self.session.__aenter__()
 
     async def close(self):
         """Closes the connection."""
         if self._session_owner and self.session:
             await self.session.close()
-            self._session_owner = False
             self.session = None
 
     def _build_ssl_config(self, cert, verify):
@@ -324,6 +333,13 @@ async def send(
                 )
                 if not stream_response:
                     await response.load_body()
+        except AttributeError as err:
+            if self.session is None:
+                raise ValueError(
+                    "No session available for request. "
+                    "Please report this issue to https://github.com/Azure/azure-sdk-for-python/issues."
+                ) from err
+            raise
         except aiohttp.client_exceptions.ClientResponseError as err:
             raise ServiceResponseError(err, error=err) from err
         except asyncio.TimeoutError as err:

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

@@ -24,7 +24,7 @@
 #
 # --------------------------------------------------------------------------
 import logging
-from typing import Iterator, Optional, Union, TypeVar, overload, cast, TYPE_CHECKING, MutableMapping
+from typing import Iterator, Optional, Union, TypeVar, overload, TYPE_CHECKING, MutableMapping
 from urllib3.util.retry import Retry
 from urllib3.exceptions import (
     DecodeError as CoreDecodeError,
@@ -250,6 +250,8 @@ def __init__(self, **kwargs) -> None:
             raise ValueError("session_owner cannot be False if no session is provided")
         self.connection_config = ConnectionConfiguration(**kwargs)
         self._use_env_settings = kwargs.pop("use_env_settings", True)
+        # See https://github.com/Azure/azure-sdk-for-python/issues/25640 to understand why we track this
+        self._has_been_opened = False
 
     def __enter__(self) -> "RequestsTransport":
         self.open()
@@ -272,16 +274,23 @@ def _init_session(self, session: requests.Session) -> None:
             session.mount(p, adapter)
 
     def open(self):
-        if not self.session and self._session_owner:
-            self.session = requests.Session()
-            self._init_session(self.session)
-        # pyright has trouble to understand that self.session is not None, since we raised at worst in the init
-        self.session = cast(requests.Session, self.session)
+        if self._has_been_opened and not self.session:
+            raise ValueError(
+                "HTTP transport has already been closed. "
+                "You may check if you're calling a function outside of the `with` of your client creation, "
+                "or if you called `close()` on your client already."
+            )
+        if not self.session:
+            if self._session_owner:
+                self.session = requests.Session()
+                self._init_session(self.session)
+            else:
+                raise ValueError("session_owner cannot be False and no session is available")
+        self._has_been_opened = True
 
     def close(self):
         if self._session_owner and self.session:
             self.session.close()
-            self._session_owner = False
             self.session = None
 
     @overload
@@ -312,7 +321,7 @@ def send(
         :keyword MutableMapping proxies: will define the proxy to use. Proxy is a dict (protocol, url)
         """
 
-    def send(
+    def send(  # pylint: disable=too-many-statements
         self,
         request: Union[HttpRequest, "RestHttpRequest"],
         *,
@@ -358,6 +367,13 @@ def send(
             )
             response.raw.enforce_content_length = True
 
+        except AttributeError as err:
+            if self.session is None:
+                raise ValueError(
+                    "No session available for request. "
+                    "Please report this issue to https://github.com/Azure/azure-sdk-for-python/issues."
+                ) from err
+            raise
         except (
             NewConnectionError,
             ConnectTimeoutError,

sdk/core/azure-core/tests/async_tests/test_basic_transport_async.py

@@ -993,3 +993,48 @@ async def test_aiohttp_errors():
     generator = AioHttpStreamDownloadGenerator(None, response)
     with pytest.raises(ServiceResponseError):
         await generator.__anext__()
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+@pytest.mark.asyncio
+async def test_already_close_with_with(caplog, port, http_request):
+    transport = AioHttpTransport()
+
+    request = http_request("GET", "http://localhost:{}/basic/string".format(port))
+
+    async with AsyncPipeline(transport) as pipeline:
+        await pipeline.run(request)
+
+    # This is now closed, new requests should fail
+    with pytest.raises(ValueError) as err:
+        await transport.send(request)
+    assert "HTTP transport has already been closed." in str(err)
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+@pytest.mark.asyncio
+async def test_already_close_manually(caplog, port, http_request):
+    transport = AioHttpTransport()
+
+    request = http_request("GET", "http://localhost:{}/basic/string".format(port))
+
+    await transport.send(request)
+    await transport.close()
+
+    # This is now closed, new requests should fail
+    with pytest.raises(ValueError) as err:
+        await transport.send(request)
+    assert "HTTP transport has already been closed." in str(err)
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+@pytest.mark.asyncio
+async def test_close_too_soon_works_fine(caplog, port, http_request):
+    transport = AioHttpTransport()
+
+    request = http_request("GET", "http://localhost:{}/basic/string".format(port))
+
+    await transport.close()
+    result = await transport.send(request)
+
+    assert result  # No exception is good enough here

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

@@ -1280,3 +1280,45 @@ def test_conflict_timeout(caplog, port, http_request):
     with pytest.raises(ValueError):
         with Pipeline(transport) as pipeline:
             pipeline.run(request, connection_timeout=(100, 100), read_timeout=100)
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_already_close_with_with(caplog, port, http_request):
+    transport = RequestsTransport()
+
+    request = http_request("GET", "http://localhost:{}/basic/string".format(port))
+
+    with Pipeline(transport) as pipeline:
+        pipeline.run(request)
+
+    # This is now closed, new requests should fail
+    with pytest.raises(ValueError) as err:
+        transport.send(request)
+    assert "HTTP transport has already been closed." in str(err)
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_already_close_manually(caplog, port, http_request):
+    transport = RequestsTransport()
+
+    request = http_request("GET", "http://localhost:{}/basic/string".format(port))
+
+    transport.send(request)
+    transport.close()
+
+    # This is now closed, new requests should fail
+    with pytest.raises(ValueError) as err:
+        transport.send(request)
+    assert "HTTP transport has already been closed." in str(err)
+
+
+@pytest.mark.parametrize("http_request", HTTP_REQUESTS)
+def test_close_too_soon_works_fine(caplog, port, http_request):
+    transport = RequestsTransport()
+
+    request = http_request("GET", "http://localhost:{}/basic/string".format(port))
+
+    transport.close()  # Never opened, should work fine
+    result = transport.send(request)
+
+    assert result  # No exception is good enough here