[dbt-snowflake] Expose ocsp_fail_open as a parameter

dbt-snowflake/.changes/unreleased/Features-20250205-213823.yaml

@@ -0,0 +1,6 @@
+kind: Features
+body: Support ocsp_fail_open setting in dbt-snowflake
+time: 2025-02-05T21:38:23.389392Z
+custom:
+    Author: jcrobak
+    Issue: "793"

dbt-snowflake/src/dbt/adapters/snowflake/connections.py

@@ -112,6 +112,7 @@ class SnowflakeCredentials(Credentials):
     insecure_mode: Optional[bool] = False
     # this needs to default to `None` so that we can tell if the user set it; see `__post_init__()`
     reuse_connections: Optional[bool] = None
+    ocsp_fail_open: Optional[bool] = None
 
     def __post_init__(self):
         if self.authenticator != "oauth" and (self.oauth_client_secret or self.oauth_client_id):
@@ -179,6 +180,7 @@ def _connection_keys(self):
             "retry_all",
             "insecure_mode",
             "reuse_connections",
+            "ocsp_fail_open",
         )
 
     def auth_args(self):
@@ -377,6 +379,11 @@ def connect():
             rec_mode = get_record_mode_from_env()
             handle = None
             if rec_mode != RecorderMode.REPLAY:
+                ocsp_fail_open_arg = (
+                    {"ocsp_fail_open": creds.ocsp_fail_open}
+                    if creds.ocsp_fail_open is not None
+                    else {}
+                )
                 handle = snowflake.connector.connect(
                     account=creds.account,
                     database=creds.database,
@@ -389,6 +396,7 @@ def connect():
                     insecure_mode=creds.insecure_mode,
                     session_parameters=session_parameters,
                     **creds.auth_args(),
+                    **ocsp_fail_open_arg,
                 )
 
             if rec_mode is not None:

dbt-snowflake/tests/unit/test_snowflake_adapter.py

@@ -718,6 +718,34 @@ def test_authenticator_private_key_string_authentication_no_passphrase(
             ]
         )
 
+    def test_set_ocsp_fail_open(self):
+        self.config.credentials = self.config.credentials.replace(ocsp_fail_open=False)
+        self.adapter = SnowflakeAdapter(self.config, get_context("spawn"))
+        conn = self.adapter.connections.set_connection_name(name="new_connection_with_new_config")
+
+        self.snowflake.assert_not_called()
+        conn.handle
+        self.snowflake.assert_has_calls(
+            [
+                mock.call(
+                    account="test-account",
+                    autocommit=True,
+                    client_session_keep_alive=False,
+                    database="test_database",
+                    role=None,
+                    schema="public",
+                    user="test_user",
+                    warehouse="test_warehouse",
+                    private_key=None,
+                    application="dbt",
+                    insecure_mode=False,
+                    session_parameters={},
+                    reuse_connections=True,
+                    ocsp_fail_open=False,
+                )
+            ]
+        )
+
 
 class TestSnowflakeAdapterConversions(TestAdapterConversions):
     def test_convert_text_type(self):