try cursor's solution

Author: sfc-gh-pczajka

test/unit/aio/test_programmatic_access_token_async.py

@@ -4,8 +4,10 @@
 
 from __future__ import annotations
 
+import os
 import pathlib
 from typing import Any, Generator
+from unittest.mock import patch
 
 import pytest
 
@@ -51,17 +53,24 @@ async def test_valid_pat_async(wiremock_client: WiremockClient) -> None:
         wiremock_generic_data_dir / "snowflake_disconnect_successful.json"
     )
 
-    connection = SnowflakeConnection(
-        user="testUser",
-        authenticator=PROGRAMMATIC_ACCESS_TOKEN,
-        token="some PAT",
-        account="testAccount",
-        protocol="http",
-        host=wiremock_client.wiremock_host,
-        port=wiremock_client.wiremock_http_port,
-    )
-    await connection.connect()
-    await connection.close()
+    # Ensure SF_PARTNER is not set so async connector uses default "AsyncioPythonConnector"
+    with patch.dict(os.environ, {}, clear=False):
+        # Remove SF_PARTNER if it exists
+        if "SF_PARTNER" in os.environ:
+            del os.environ["SF_PARTNER"]
+
+        connection = SnowflakeConnection(
+            user="testUser",
+            authenticator=PROGRAMMATIC_ACCESS_TOKEN,
+            token="some PAT",
+            account="testAccount",
+            protocol="http",
+            host=wiremock_client.wiremock_host,
+            port=wiremock_client.wiremock_http_port,
+            autocommit=True,  # Enable autocommit to prevent COMMIT calls during cleanup
+        )
+        await connection.connect()
+        await connection.close()
 
 
 @pytest.mark.skipolddriver
@@ -77,19 +86,27 @@ async def test_invalid_pat_async(wiremock_client: WiremockClient) -> None:
     )
     wiremock_client.import_mapping(wiremock_data_dir / "invalid_token_async.json")
 
-    with pytest.raises(snowflake.connector.errors.DatabaseError) as execinfo:
-        connection = SnowflakeConnection(
-            user="testUser",
-            authenticator=PROGRAMMATIC_ACCESS_TOKEN,
-            token="some PAT",
-            account="testAccount",
-            protocol="http",
-            host=wiremock_client.wiremock_host,
-            port=wiremock_client.wiremock_http_port,
-        )
-        await connection.connect()
-
-    assert str(execinfo.value).endswith("Programmatic access token is invalid.")
+    # Ensure SF_PARTNER is not set so async connector uses default "AsyncioPythonConnector"
+    with patch.dict(os.environ, {}, clear=False):
+        # Remove SF_PARTNER if it exists
+        if "SF_PARTNER" in os.environ:
+            del os.environ["SF_PARTNER"]
+
+        with pytest.raises(snowflake.connector.errors.DatabaseError) as execinfo:
+            connection = SnowflakeConnection(
+                user="testUser",
+                authenticator=PROGRAMMATIC_ACCESS_TOKEN,
+                token="some PAT",
+                account="testAccount",
+                protocol="http",
+                host=wiremock_client.wiremock_host,
+                port=wiremock_client.wiremock_http_port,
+                autocommit=True,  # Enable autocommit to prevent COMMIT calls during cleanup
+            )
+            await connection.connect()
+        await connection.close()
+
+        assert str(execinfo.value).endswith("Programmatic access token is invalid.")
 
 
 @pytest.mark.skipolddriver
@@ -119,15 +136,22 @@ async def test_pat_as_password_async(wiremock_client: WiremockClient) -> None:
         wiremock_generic_data_dir / "snowflake_disconnect_successful.json"
     )
 
-    connection = SnowflakeConnection(
-        user="testUser",
-        authenticator=PROGRAMMATIC_ACCESS_TOKEN,
-        token=None,
-        password="some PAT",
-        account="testAccount",
-        protocol="http",
-        host=wiremock_client.wiremock_host,
-        port=wiremock_client.wiremock_http_port,
-    )
-    await connection.connect()
-    await connection.close()
+    # Ensure SF_PARTNER is not set so async connector uses default "AsyncioPythonConnector"
+    with patch.dict(os.environ, {}, clear=False):
+        # Remove SF_PARTNER if it exists
+        if "SF_PARTNER" in os.environ:
+            del os.environ["SF_PARTNER"]
+
+        connection = SnowflakeConnection(
+            user="testUser",
+            authenticator=PROGRAMMATIC_ACCESS_TOKEN,
+            token=None,
+            password="some PAT",
+            account="testAccount",
+            protocol="http",
+            host=wiremock_client.wiremock_host,
+            port=wiremock_client.wiremock_http_port,
+            autocommit=True,  # Enable autocommit to prevent COMMIT calls during cleanup
+        )
+        await connection.connect()
+        await connection.close()