Add test_connection method for SSHHook

airflow/providers/ssh/hooks/ssh.py

@@ -527,3 +527,12 @@ def exec_ssh_client_command(
         exit_status = stdout.channel.recv_exit_status()
 
         return exit_status, agg_stdout, agg_stderr
+
+    def test_connection(self) -> tuple[bool, str]:
+        """Test the ssh connection by execute remote bash commands"""
+        try:
+            with self.get_conn() as conn:
+                conn.exec_command("pwd")
+            return True, "Connection successfully tested"
+        except Exception as e:
+            return False, str(e)

tests/providers/ssh/hooks/test_ssh.py

@@ -955,3 +955,15 @@ def test_ssh_connection_with_no_host_key_check_true_and_allow_host_key_changes_f
                     ssh_mock.return_value.set_missing_host_key_policy.call_args[0][0], paramiko.AutoAddPolicy
                 )
                 assert ssh_mock.return_value.load_host_keys.called is False
+
+    def test_connection_success(self):
+        hook = SSHHook(ssh_conn_id="ssh_default")
+        status, msg = hook.test_connection()
+        assert status is True
+        assert msg == "Connection successfully tested"
+    def test_connection_failure(self):
+        hook = SSHHook(ssh_conn_id="ssh_default")
+        hook.get_conn = mock.MagicMock(name="mock_conn", side_effect=Exception("Test failure case"))
+        status, msg = hook.test_connection()
+        assert status is False
+        assert msg == "Test failure case"