Add test_connection to SalesforceHook

airflow/providers/salesforce/hooks/salesforce.py

@@ -391,3 +391,15 @@ def object_to_df(
             df["time_fetched_from_salesforce"] = fetched_time
 
         return df
+
+    def test_connection(self):
+        """Test the Salesforce connectivity"""
+        try:
+            self.describe_object("Account")
+            status = True
+            message = "Connection successfully tested"
+        except Exception as e:
+            status = False
+            message = str(e)
+
+        return status, message

tests/providers/salesforce/hooks/test_salesforce.py

@@ -515,3 +515,23 @@ def test_backcompat_prefix_both_prefers_short(self, mock_client):
                 username=None,
                 version="52.0",
             )
+
+    @patch(
+        "airflow.providers.salesforce.hooks.salesforce.SalesforceHook.describe_object",
+        return_value={"fields": [{"name": "field_1"}, {"name": "field_2"}]},
+    )
+    def test_connection_success(self, mock_describe_object):
+        hook = SalesforceHook("my_conn")
+        status, msg = hook.test_connection()
+        assert status is True
+        assert msg == "Connection successfully tested"
+
+    @patch(
+        "airflow.providers.salesforce.hooks.salesforce.SalesforceHook.describe_object",
+        side_effect=Exception("Test"),
+    )
+    def test_connection_failure(self, mock_describe_object):
+        hook = SalesforceHook("my_conn")
+        status, msg = hook.test_connection()
+        assert status is False
+        assert msg == "Test"