add raising mechanism + test

Author: cepedus

aiobotocore/httpsession.py

@@ -3,6 +3,7 @@
 import io
 import os
 import socket
+from threading import Lock
 from typing import Optional
 
 import aiohttp  # lgtm [py/import-and-import-from]
@@ -44,6 +45,8 @@
 from ._constants import DEFAULT_KEEPALIVE_TIMEOUT
 from ._endpoint_helpers import _IOBaseWrapper, _text
 
+_session_lock = Lock()
+
 
 class AIOHTTPSession:
     def __init__(
@@ -99,14 +102,30 @@ def __init__(
         # request so don't need proxy manager
 
     async def __aenter__(self):
-        assert not self._sessions
+        with _session_lock:
+            self._aexited = False
 
+        assert not self._sessions
         return self
 
     async def __aexit__(self, exc_type, exc_val, exc_tb):
         self._sessions.clear()
         await self._exit_stack.aclose()
 
+        with _session_lock:
+            self._aexited = True
+
+    @property
+    def _can_create_session(self) -> bool:
+        """Check if a new client session can be created.
+
+        This is true for two cases:
+        - We are inside the context manager of the class.
+        - We have never attempted to enter the context manager.
+        """
+        with _session_lock:
+            return not getattr(self, '_aexited', False)
+
     def _get_ssl_context(self):
         return create_urllib3_context()
 
@@ -173,6 +192,11 @@ def _create_connector(self, proxy_url):
 
     async def _get_session(self, proxy_url):
         if not (session := self._sessions.get(proxy_url)):
+            if not self._can_create_session:
+                raise RuntimeError(
+                    f'Cannot create a new {aiohttp.ClientSession.__name__} '
+                    'when context manager was exited.'
+                )
             connector = self._create_connector(proxy_url)
             self._sessions[proxy_url] = (
                 session

tests/botocore_tests/unit/test_session.py

@@ -152,6 +152,21 @@ async def test_can_create_client(self, session):
         async with sts_client_context as sts_client:
             assert isinstance(sts_client, client.AioBaseClient)
 
+    async def test_cannot_create_client_sessions_outside_context(
+        self, session
+    ):
+        s3_client_context = session.create_client('s3', 'us-west-2')
+
+        async with s3_client_context as s3_client:
+            pass
+
+        with pytest.raises(
+            botocore.exceptions.HTTPClientError,
+            match='Cannot create a new ClientSession when context manager was'
+            ' exited',
+        ):
+            await s3_client.list_buckets()
+
     async def test_credential_provider_not_called_when_creds_provided(
         self, session
     ):