[async] Fixed #2429 and #2568:

conn-rest-conn -> conn and made http_config passed from async to base_auth_data.
Fix - limited outgoing requests from tests
Fixed not awaited coroutine error
Fixed old mistakes in _okta.py - conn.rest.connection -> conn
Fixed errors with wif tests
Fixed errors with okta auth step4_negative test case
Fixed errors with sessionManager runtime import and None value
Fixed errors with no session manager in workload_identity in tests
Fixed Connection closed issue

Author: sfc-gh-fpawlowski

src/snowflake/connector/aio/_network.py

@@ -849,7 +849,7 @@ async def _request_exec(
             ) from err
 
     @contextlib.asynccontextmanager
-    async def _use_session(
+    async def use_session(
         self, url: str | None = None
     ) -> AsyncGenerator[aiohttp.ClientSession]:
         async with self._session_manager.use_session(url) as session:

src/snowflake/connector/aio/_s3_storage_client.py

@@ -425,8 +425,27 @@ async def _has_expired_token(self, response: aiohttp.ClientResponse) -> bool:
         """
         if response.status != 400:
             return False
-        message = await response.text()
+        # Read body once; avoid a second read which can raise RuntimeError("Connection closed.")
+        try:
+            message = await response.text()
+        except RuntimeError as e:
+            logger.debug(
+                "S3 token-expiry check: failed to read error body, treating as not expired. error=%s",
+                type(e),
+            )
+            return False
         if not message:
+            logger.debug(
+                "S3 token-expiry check: empty error body, treating as not expired"
+            )
+            return False
+        try:
+            err = ET.fromstring(message)
+        except ET.ParseError:
+            logger.debug(
+                "S3 token-expiry check: non-XML error body (len=%d), treating as not expired.",
+                len(message),
+            )
             return False
-        err = ET.fromstring(await response.read())
-        return err.find("Code").text == EXPIRED_TOKEN
+        code = err.find("Code")
+        return code is not None and code.text == EXPIRED_TOKEN

src/snowflake/connector/aio/_wif_util.py

@@ -4,7 +4,6 @@
 import logging
 import os
 from base64 import b64encode
-from typing import TYPE_CHECKING
 
 import aioboto3
 from aiobotocore.utils import AioInstanceMetadataRegionFetcher
@@ -22,9 +21,7 @@
     extract_iss_and_sub_without_signature_verification,
     get_aws_sts_hostname,
 )
-
-if TYPE_CHECKING:
-    from ._session_manager import SessionManager
+from ._session_manager import SessionManager
 
 logger = logging.getLogger(__name__)
 
@@ -175,6 +172,9 @@ async def create_attestation(
     If an explicit entra_resource was provided to the connector, this will be used. Otherwise, the default Snowflake Entra resource will be used.
     """
     entra_resource = entra_resource or DEFAULT_ENTRA_SNOWFLAKE_RESOURCE
+    session_manager = (
+        session_manager.clone() if session_manager else SessionManager(use_pooling=True)
+    )
 
     if provider == AttestationProvider.AWS:
         return await create_aws_attestation()

src/snowflake/connector/aio/auth/_auth.py

@@ -103,7 +103,7 @@ async def authenticate(
             self._rest._connection._network_timeout,
             self._rest._connection._socket_timeout,
             self._rest._connection._platform_detection_timeout_seconds,
-            session_manager=self._rest.session_manager.clone(use_pooling=False),
+            http_config=self._rest.session_manager.config,  # AioHttpConfig extends BaseHttpConfig
         )
 
         body = copy.deepcopy(body_template)

src/snowflake/connector/aio/auth/_okta.py

@@ -123,6 +123,7 @@ async def _step1(
             conn._ocsp_mode(),
             conn.login_timeout,
             conn._network_timeout,
+            http_config=conn._session_manager.config,  # AioHttpConfig extends BaseHttpConfig
         )
 
         body["data"]["AUTHENTICATOR"] = authenticator
@@ -131,12 +132,12 @@ async def _step1(
             account,
             authenticator,
         )
-        ret = await conn._rest._post_request(
+        ret = await conn.rest._post_request(
             url,
             headers,
             json.dumps(body),
-            timeout=conn._rest._connection.login_timeout,
-            socket_timeout=conn._rest._connection.login_timeout,
+            timeout=conn.login_timeout,
+            socket_timeout=conn.login_timeout,
         )
 
         if not ret["success"]:
@@ -171,19 +172,19 @@ async def _step3(
             "username": user,
             "password": password,
         }
-        ret = await conn._rest.fetch(
+        ret = await conn.rest.fetch(
             "post",
             token_url,
             headers,
             data=json.dumps(data),
-            timeout=conn._rest._connection.login_timeout,
-            socket_timeout=conn._rest._connection.login_timeout,
+            timeout=conn.login_timeout,
+            socket_timeout=conn.login_timeout,
             catch_okta_unauthorized_error=True,
         )
         one_time_token = ret.get("sessionToken", ret.get("cookieToken"))
         if not one_time_token:
             Error.errorhandler_wrapper(
-                conn._rest._connection,
+                conn,
                 None,
                 DatabaseError,
                 {
@@ -221,7 +222,7 @@ async def _step4(
                     HTTP_HEADER_ACCEPT: "*/*",
                 }
                 remaining_timeout = timeout_time - time.time() if timeout_time else None
-                response_html = await conn._rest.fetch(
+                response_html = await conn.rest.fetch(
                     "get",
                     sso_url,
                     headers,

src/snowflake/connector/aio/auth/_webbrowser.py

@@ -365,13 +365,13 @@ async def _get_sso_url(
         body = Auth.base_auth_data(
             user,
             account,
-            conn._rest._connection.application,
-            conn._rest._connection._internal_application_name,
-            conn._rest._connection._internal_application_version,
-            conn._rest._connection._ocsp_mode(),
-            conn._rest._connection.login_timeout,
-            conn._rest._connection._network_timeout,
-            session_manager=conn.rest.session_manager.clone(use_pooling=False),
+            conn.application,
+            conn._internal_application_name,
+            conn._internal_application_version,
+            conn._ocsp_mode(),
+            conn.login_timeout,
+            conn._network_timeout,
+            http_config=conn._session_manager.config,  # AioHttpConfig extends BaseHttpConfig
         )
 
         body["data"]["AUTHENTICATOR"] = authenticator
@@ -383,8 +383,8 @@ async def _get_sso_url(
             url,
             headers,
             json.dumps(body),
-            timeout=conn._rest._connection.login_timeout,
-            socket_timeout=conn._rest._connection.login_timeout,
+            timeout=conn.login_timeout,
+            socket_timeout=conn.login_timeout,
         )
         if not ret["success"]:
             await self._handle_failure(conn=conn, ret=ret)

src/snowflake/connector/auth/_auth.py

@@ -53,7 +53,8 @@
     ReauthenticationRequest,
 )
 from ..platform_detection import detect_platforms
-from ..session_manager import SessionManager
+from ..session_manager import BaseHttpConfig, HttpConfig
+from ..session_manager import SessionManager as SyncSessionManager
 from ..sqlstate import SQLSTATE_CONNECTION_WAS_NOT_ESTABLISHED
 from ..token_cache import TokenCache, TokenKey, TokenType
 from ..version import VERSION
@@ -104,8 +105,17 @@ def base_auth_data(
         network_timeout: int | None = None,
         socket_timeout: int | None = None,
         platform_detection_timeout_seconds: float | None = None,
-        session_manager: SessionManager | None = None,
+        session_manager: SyncSessionManager | None = None,
+        http_config: BaseHttpConfig | None = None,
     ):
+        # Create sync SessionManager for platform detection if config is provided
+        # Platform detection runs in threads and uses sync SessionManager
+        if http_config is not None and session_manager is None:
+            # Extract base fields (automatically excludes subclass-specific fields)
+            # Note: It won't be possible to pass adapter_factory from outer async-code to this part of code
+            sync_config = HttpConfig(**http_config.to_base_dict())
+            session_manager = SyncSessionManager(config=sync_config)
+
         return {
             "data": {
                 "CLIENT_APP_ID": internal_application_name,

src/snowflake/connector/platform_detection.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import logging
 import os
 import re
 from concurrent.futures.thread import ThreadPoolExecutor
@@ -13,6 +14,8 @@
 from .session_manager import SessionManager
 from .vendored.requests import RequestException, Timeout
 
+logger = logging.getLogger(__name__)
+
 
 class _DetectionState(Enum):
     """Internal enum to represent the detection state of a platform."""
@@ -399,6 +402,9 @@ def detect_platforms(
 
         if session_manager is None:
             # This should never happen - we expect session manager to be passed from the outer scope
+            logger.debug(
+                "No session manager provided. HTTP settings may not be preserved. Using default."
+            )
             session_manager = SessionManager(use_pooling=False)
 
         # Run environment-only checks synchronously (no network calls, no threading overhead)

src/snowflake/connector/session_manager.py

@@ -6,7 +6,7 @@
 import functools
 import itertools
 import logging
-from dataclasses import dataclass, field, replace
+from dataclasses import asdict, dataclass, field, fields, replace
 from typing import TYPE_CHECKING, Any, Callable, Generator, Generic, Mapping, TypeVar
 
 from .compat import urlparse
@@ -117,6 +117,11 @@ def copy_with(self, **overrides: Any) -> BaseHttpConfig:
         """Return a new config with overrides applied."""
         return replace(self, **overrides)
 
+    def to_base_dict(self) -> dict[str, Any]:
+        """Extract only BaseHttpConfig fields as a dict, excluding subclass-specific fields."""
+        base_field_names = {f.name for f in fields(BaseHttpConfig)}
+        return {k: v for k, v in asdict(self).items() if k in base_field_names}
+
 
 @dataclass(frozen=True)
 class HttpConfig(BaseHttpConfig):

test/integ/aio_it/pandas_it/test_arrow_pandas_async.py

@@ -1402,8 +1402,8 @@ async def test_sessions_used(conn_cnx, fetch_fn_name, pass_connection):
 
             # check that sessions are used when connection is supplied
             with mock.patch(
-                "snowflake.connector.aio._network.SnowflakeRestful._use_session",
-                side_effect=cnx._rest._use_session,
+                "snowflake.connector.aio._network.SnowflakeRestful.use_session",
+                side_effect=cnx._rest.use_session,
             ) as get_session_mock:
                 await fetch_fn(connection=connection)
                 assert get_session_mock.call_count == (1 if pass_connection else 0)