add snowflake sso

data_diff/dbt.py

@@ -480,8 +480,6 @@ def set_connection(self):
         credentials, conn_type = self._get_connection_creds()
 
         if conn_type == "snowflake":
-            if credentials.get("authenticator") is not None:
-                raise Exception("Federated authentication is not yet supported for Snowflake.")
             conn_info = {
                 "driver": conn_type,
                 "user": credentials.get("user"),
@@ -499,10 +497,13 @@ def set_connection(self):
                     raise Exception("Cannot use password and key at the same time")
                 conn_info["key"] = credentials.get("private_key_path")
                 conn_info["private_key_passphrase"] = credentials.get("private_key_passphrase")
+            elif credentials.get("authenticator") is not None:
+                conn_info["authenticator"] = credentials.get("authenticator")
+                conn_info["password"] = credentials.get("password")
             elif credentials.get("password") is not None:
                 conn_info["password"] = credentials.get("password")
             else:
-                raise Exception("Password or key authentication not provided.")
+                raise Exception("Snowflake: unsupported auth method")
         elif conn_type == "bigquery":
             method = credentials.get("method")
             # there are many connection types https://docs.getdbt.com/reference/warehouse-setups/bigquery-setup#oauth-via-gcloud

tests/test_dbt.py

@@ -199,10 +199,12 @@ def test_set_connection_snowflake_authenticator(self):
         mock_self = Mock()
         mock_self._get_connection_creds.return_value = (expected_credentials, expected_driver)
 
-        with self.assertRaises(Exception):
-            DbtParser.set_connection(mock_self)
+        DbtParser.set_connection(mock_self)
 
-        self.assertNotIsInstance(mock_self.connection, dict)
+        self.assertIsInstance(mock_self.connection, dict)
+        self.assertEqual(mock_self.connection.get("driver"), expected_driver)
+        self.assertEqual(mock_self.connection.get("authenticator"), expected_credentials["authenticator"])
+        self.assertEqual(mock_self.connection.get("user"), expected_credentials["user"])
 
     def test_set_connection_snowflake_key_and_password(self):
         expected_driver = "snowflake"