Check ocsp errors

src/snowflake/connector/connection.py

@@ -1545,6 +1545,8 @@ def __config(self, **kwargs):
             WORKLOAD_IDENTITY_AUTHENTICATOR,
             PROGRAMMATIC_ACCESS_TOKEN,
             PAT_WITH_EXTERNAL_SESSION,
+            OAUTH_AUTHORIZATION_CODE,
+            OAUTH_CLIENT_CREDENTIALS,
         }
 
         if not (self._master_token and self._session_token):

test/unit/aio/test_auth_oauth_auth_code_async.py

@@ -301,3 +301,51 @@ def test_mro():
     assert AuthByOauthCode.mro().index(AuthByPluginAsync) < AuthByOauthCode.mro().index(
         AuthByPluginSync
     )
+
+
+@pytest.mark.skipolddriver
+async def test_oauth_authorization_code_allows_empty_user(
+    monkeypatch, omit_oauth_urls_check
+):
+    """Test that OAUTH_AUTHORIZATION_CODE authenticator allows connection without user parameter."""
+    import snowflake.connector.aio
+    from snowflake.connector.aio._network import SnowflakeRestful
+
+    async def mock_post_request(self, url, headers, json_body, **kwargs):
+        return {
+            "success": True,
+            "message": None,
+            "data": {
+                "token": "TOKEN",
+                "masterToken": "MASTER_TOKEN",
+                "idToken": None,
+                "parameters": [{"name": "SERVICE_NAME", "value": "FAKE_SERVICE_NAME"}],
+            },
+        }
+
+    monkeypatch.setattr(SnowflakeRestful, "_post_request", mock_post_request)
+
+    # Mock the OAuth authorization flow to avoid opening browser and starting HTTP server
+    # Note: This must be a sync function (not async) because it's called from the sync
+    # parent class's prepare() method which calls _request_tokens() without await
+    def mock_request_tokens(self, **kwargs):
+        # Simulate successful token retrieval
+        return ("mock_access_token", "mock_refresh_token")
+
+    monkeypatch.setattr(AuthByOauthCode, "_request_tokens", mock_request_tokens)
+
+    # Test connection without user parameter - should succeed
+    conn = snowflake.connector.aio.SnowflakeConnection(
+        account="testaccount",
+        authenticator="OAUTH_AUTHORIZATION_CODE",
+        oauth_client_id="test_client_id",
+        oauth_client_secret="test_client_secret",
+    )
+
+    await conn.connect()
+
+    # Verify that the connection was successful
+    assert conn is not None
+    assert isinstance(conn.auth_class, AuthByOauthCode)
+
+    await conn.close()

test/unit/aio/test_auth_oauth_credentials_async.py

@@ -144,6 +144,56 @@ def mock_get_request_token_response(self, connection, fields):
     await conn.close()
 
 
+@pytest.mark.skipolddriver
+async def test_oauth_client_credentials_allows_empty_user(monkeypatch):
+    """Test that OAUTH_CLIENT_CREDENTIALS authenticator allows connection without user parameter."""
+    import snowflake.connector.aio
+
+    async def mock_post_request(self, url, headers, json_body, **kwargs):
+        return {
+            "success": True,
+            "message": None,
+            "data": {
+                "token": "TOKEN",
+                "masterToken": "MASTER_TOKEN",
+                "idToken": None,
+                "parameters": [{"name": "SERVICE_NAME", "value": "FAKE_SERVICE_NAME"}],
+            },
+        }
+
+    monkeypatch.setattr(
+        snowflake.connector.aio._network.SnowflakeRestful,
+        "_post_request",
+        mock_post_request,
+    )
+
+    # Mock the OAuth client credentials token request to avoid making HTTP requests
+    def mock_get_request_token_response(self, connection, fields):
+        return ("mocked_token_response", None)
+
+    monkeypatch.setattr(
+        AuthByOauthCredentials,
+        "_get_request_token_response",
+        mock_get_request_token_response,
+    )
+
+    # Test connection without user parameter - should succeed
+    conn = snowflake.connector.aio.SnowflakeConnection(
+        account="testaccount",
+        authenticator="OAUTH_CLIENT_CREDENTIALS",
+        oauth_client_id="test_client_id",
+        oauth_client_secret="test_client_secret",
+    )
+
+    await conn.connect()
+
+    # Verify that the connection was successful
+    assert conn is not None
+    assert isinstance(conn.auth_class, AuthByOauthCredentials)
+
+    await conn.close()
+
+
 async def test_oauth_credentials_missing_client_id_raises_error():
     """Test that missing client_id raises a ProgrammingError."""
     with pytest.raises(ProgrammingError) as excinfo:

test/unit/test_auth_oauth_auth_code.py

@@ -285,3 +285,46 @@ def mock_request_tokens(self, **kwargs):
     assert isinstance(conn.auth_class, AuthByOauthCode)
 
     conn.close()
+
+
+@pytest.mark.skipolddriver
+def test_oauth_authorization_code_allows_empty_user(monkeypatch, omit_oauth_urls_check):
+    """Test that OAUTH_AUTHORIZATION_CODE authenticator allows connection without user parameter."""
+    import snowflake.connector
+
+    def mock_post_request(self, url, headers, json_body, **kwargs):
+        return {
+            "success": True,
+            "message": None,
+            "data": {
+                "token": "TOKEN",
+                "masterToken": "MASTER_TOKEN",
+                "idToken": None,
+                "parameters": [{"name": "SERVICE_NAME", "value": "FAKE_SERVICE_NAME"}],
+            },
+        }
+
+    monkeypatch.setattr(
+        snowflake.connector.network.SnowflakeRestful, "_post_request", mock_post_request
+    )
+
+    # Mock the OAuth authorization flow to avoid opening browser and starting HTTP server
+    def mock_request_tokens(self, **kwargs):
+        # Simulate successful token retrieval
+        return ("mock_access_token", "mock_refresh_token")
+
+    monkeypatch.setattr(AuthByOauthCode, "_request_tokens", mock_request_tokens)
+
+    # Test connection without user parameter - should succeed
+    conn = snowflake.connector.connect(
+        account="testaccount",
+        authenticator="OAUTH_AUTHORIZATION_CODE",
+        oauth_client_id="test_client_id",
+        oauth_client_secret="test_client_secret",
+    )
+
+    # Verify that the connection was successful
+    assert conn is not None
+    assert isinstance(conn.auth_class, AuthByOauthCode)
+
+    conn.close()

test/unit/test_auth_oauth_credentials.py

@@ -137,6 +137,53 @@ def mock_get_request_token_response(self, connection, fields):
     conn.close()
 
 
+@pytest.mark.skipolddriver
+def test_oauth_client_credentials_allows_empty_user(monkeypatch):
+    """Test that OAUTH_CLIENT_CREDENTIALS authenticator allows connection without user parameter."""
+    import snowflake.connector
+
+    def mock_post_request(request, url, headers, json_body, **kwargs):
+        return {
+            "success": True,
+            "message": None,
+            "data": {
+                "token": "TOKEN",
+                "masterToken": "MASTER_TOKEN",
+                "idToken": None,
+                "parameters": [{"name": "SERVICE_NAME", "value": "FAKE_SERVICE_NAME"}],
+            },
+        }
+
+    monkeypatch.setattr(
+        "snowflake.connector.network.SnowflakeRestful._post_request",
+        mock_post_request,
+    )
+
+    # Mock the OAuth client credentials token request to avoid making HTTP requests
+    def mock_get_request_token_response(self, connection, fields):
+        return ("mocked_token_response", None)
+
+    monkeypatch.setattr(
+        AuthByOauthCredentials,
+        "_get_request_token_response",
+        mock_get_request_token_response,
+    )
+
+    # Test connection without user parameter - should succeed
+    conn = snowflake.connector.connect(
+        account="testaccount",
+        authenticator="OAUTH_CLIENT_CREDENTIALS",
+        oauth_client_id="test_client_id",
+        oauth_client_secret="test_client_secret",
+    )
+
+    # Verify that the connection was successful
+    assert conn is not None
+    assert isinstance(conn.auth_class, AuthByOauthCredentials)
+
+    conn.close()
+
+
 def test_oauth_credentials_missing_client_id_raises_error():
     """Test that missing client_id raises a ProgrammingError."""
     with pytest.raises(ProgrammingError) as excinfo: