Bugfix: SFTPHook does not respect ssh_conn_id arg (#20756)

closes #20735

airflow/providers/sftp/hooks/sftp.py

@@ -74,19 +74,18 @@ def get_ui_field_behaviour() -> Dict:
     def __init__(
         self,
         ssh_conn_id: Optional[str] = 'sftp_default',
-        ftp_conn_id: Optional[str] = 'sftp_default',
         *args,
         **kwargs,
     ) -> None:
-
+        ftp_conn_id = kwargs.pop('ftp_conn_id', None)
         if ftp_conn_id:
             warnings.warn(
-                'Parameter `ftp_conn_id` is deprecated.' 'Please use `ssh_conn_id` instead.',
+                'Parameter `ftp_conn_id` is deprecated. Please use `ssh_conn_id` instead.',
                 DeprecationWarning,
                 stacklevel=2,
             )
-            kwargs['ssh_conn_id'] = ftp_conn_id
-        self.ssh_conn_id = ssh_conn_id
+            ssh_conn_id = ftp_conn_id
+        kwargs['ssh_conn_id'] = ssh_conn_id
         super().__init__(*args, **kwargs)
 
         self.conn = None

tests/providers/sftp/hooks/test_sftp.py

@@ -334,6 +334,25 @@ def test_connection_success(self, mock_get_connection):
         assert status is True
         assert msg == 'Connection successfully tested'
 
+    @mock.patch('airflow.providers.sftp.hooks.sftp.SFTPHook.get_connection')
+    def test_deprecation_ftp_conn_id(self, mock_get_connection):
+        connection = Connection(conn_id='ftp_default', login='login', host='host')
+        mock_get_connection.return_value = connection
+        # If `ftp_conn_id` is provided, it will be used but would show a deprecation warning.
+        with self.assertWarnsRegex(DeprecationWarning, "Parameter `ftp_conn_id` is deprecated"):
+            assert SFTPHook(ftp_conn_id='ftp_default').ssh_conn_id == 'ftp_default'
+
+        # If both are provided, ftp_conn_id  will be used but would show a deprecation warning.
+        with self.assertWarnsRegex(DeprecationWarning, "Parameter `ftp_conn_id` is deprecated"):
+            assert (
+                SFTPHook(ftp_conn_id='ftp_default', ssh_conn_id='sftp_default').ssh_conn_id == 'ftp_default'
+            )
+
+        # If `ssh_conn_id` is provided, it should use it for ssh_conn_id
+        assert SFTPHook(ssh_conn_id='sftp_default').ssh_conn_id == 'sftp_default'
+        # Default is 'sftp_default
+        assert SFTPHook().ssh_conn_id == 'sftp_default'
+
     def tearDown(self):
         shutil.rmtree(os.path.join(TMP_PATH, TMP_DIR_FOR_TESTS))
         os.remove(os.path.join(TMP_PATH, TMP_FILE_FOR_TESTS))