[airflow] branch main updated: fix(providers/ftp): respect soft_fail argument when exception is raised (#34161)

[po...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 4f0e4254e4 fix(providers/ftp): respect soft_fail argument when exception is raised (#34161)
4f0e4254e4 is described below

commit 4f0e4254e4d8b0c3f14b7caf228e30f9e1577123
Author: Wei Lee <we...@gmail.com>
AuthorDate: Thu Sep 7 18:38:00 2023 +0800

    fix(providers/ftp): respect soft_fail argument when exception is raised (#34161)
---
 airflow/providers/ftp/sensors/ftp.py    |  3 +++
 tests/providers/ftp/sensors/test_ftp.py | 12 ++++++++++++
 2 files changed, 15 insertions(+)

diff --git a/airflow/providers/ftp/sensors/ftp.py b/airflow/providers/ftp/sensors/ftp.py
index 7dddbf521a..19b69fb994 100644
--- a/airflow/providers/ftp/sensors/ftp.py
+++ b/airflow/providers/ftp/sensors/ftp.py
@@ -21,6 +21,7 @@ import ftplib
 import re
 from typing import TYPE_CHECKING, Sequence
 
+from airflow.exceptions import AirflowSkipException
 from airflow.providers.ftp.hooks.ftp import FTPHook, FTPSHook
 from airflow.sensors.base import BaseSensorOperator
 
@@ -82,6 +83,8 @@ class FTPSensor(BaseSensorOperator):
                 if (error_code != 550) and (
                     self.fail_on_transient_errors or (error_code not in self.transient_errors)
                 ):
+                    if self.soft_fail:
+                        raise AirflowSkipException from e
                     raise e
 
                 return False
diff --git a/tests/providers/ftp/sensors/test_ftp.py b/tests/providers/ftp/sensors/test_ftp.py
index e79b46526a..107b232371 100644
--- a/tests/providers/ftp/sensors/test_ftp.py
+++ b/tests/providers/ftp/sensors/test_ftp.py
@@ -22,6 +22,7 @@ from unittest import mock
 
 import pytest
 
+from airflow.exceptions import AirflowSkipException
 from airflow.providers.ftp.hooks.ftp import FTPHook
 from airflow.providers.ftp.sensors.ftp import FTPSensor
 
@@ -69,6 +70,17 @@ class TestFTPSensor:
 
         assert "434" in str(ctx.value)
 
+    @mock.patch("airflow.providers.ftp.sensors.ftp.FTPHook", spec=FTPHook)
+    def test_poke_fail_on_transient_error_and_skip(self, mock_hook):
+        op = FTPSensor(path="foobar.json", ftp_conn_id="bob_ftp", task_id="test_task", soft_fail=True)
+
+        mock_hook.return_value.__enter__.return_value.get_mod_time.side_effect = error_perm(
+            "434: Host unavailable"
+        )
+
+        with pytest.raises(AirflowSkipException):
+            op.execute(None)
+
     @mock.patch("airflow.providers.ftp.sensors.ftp.FTPHook", spec=FTPHook)
     def test_poke_ignore_transient_error(self, mock_hook):
         op = FTPSensor(