Add test_connection method for Snowflake Hook (#19041)

Co-authored-by: Tzu-ping Chung <uranusjr@gmail.com>

airflow/providers/snowflake/hooks/snowflake.py

@@ -302,3 +302,11 @@ def run(self, sql: Union[str, list], autocommit: bool = False, parameters: Optio
                 conn.commit()
 
         return execution_info
+
+    def test_connection(self):
+        """Test the Snowflake connection by running a simple query."""
+        try:
+            self.run(sql="select 1")
+        except Exception as e:
+            return False, str(e)
+        return True, "Connection successfully tested"

airflow/www/static/js/connection_form.js

@@ -173,7 +173,7 @@ $(document).ready(() => {
         outObj.connection_id = this.value;
       } else if (this.value !== '' && this.name === 'port') {
         outObj[this.name] = Number(this.value);
-      } else if (this.value !== '' && this.name !== 'csrf_token') {
+      } else if (this.value !== '' && this.name !== 'csrf_token' && !this.name.match('extra__')) {
         outObj[this.name] = this.value;
       }
     });

tests/providers/snowflake/hooks/test_snowflake.py

@@ -180,6 +180,22 @@ def test_key_pair_auth_not_encrypted(self):
         params = self.db_hook._get_conn_params()
         assert 'private_key' in params
 
+    @mock.patch('airflow.providers.snowflake.hooks.snowflake.SnowflakeHook.run')
+    def test_connection_success(self, mock_run):
+        mock_run.return_value = [{'1': 1}]
+        status, msg = self.db_hook.test_connection()
+        assert status is True
+        assert msg == 'Connection successfully tested'
+
+    @mock.patch(
+        'airflow.providers.snowflake.hooks.snowflake.SnowflakeHook.run',
+        side_effect=Exception('Connection Errors'),
+    )
+    def test_connection_failure(self, mock_run):
+        status, msg = self.db_hook.test_connection()
+        assert status is False
+        assert msg == 'Connection Errors'
+
 
 """
     Testing hooks with assigning`extra_` parameters