You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "Lee-W (via GitHub)" <gi...@apache.org> on 2023/09/28 14:19:17 UTC
[GitHub] [airflow] Lee-W commented on a diff in pull request #34652: Add testcases to ensure that 'soft_fail' argument is respected when running ExternalTaskSensor
Lee-W commented on code in PR #34652:
URL: https://github.com/apache/airflow/pull/34652#discussion_r1340227292
##########
tests/sensors/test_external_task_sensor.py:
##########
@@ -838,6 +839,172 @@ def test_external_task_group_when_there_is_no_TIs(self):
ignore_ti_state=True,
)
+ @pytest.mark.parametrize(
+ "soft_fail, expected_exception, kwargs, expected_message",
+ (
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"Some of the external tasks {re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}"
+ f" in DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"The external task_group '{re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}'"
+ f" in DAG '{TEST_DAG_ID}' failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {"failed_states": [State.FAILED]},
+ f"The external DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {"failed_states": [State.FAILED]},
+ "Skipping due to soft_fail is set to True.",
+ ),
+ ),
+ )
+ @mock.patch("airflow.sensors.external_task.ExternalTaskSensor.get_count")
+ @mock.patch("airflow.sensors.external_task.ExternalTaskSensor._get_dttm_filter")
+ def test_fail_poke(
+ self, _get_dttm_filter, get_count, soft_fail, expected_exception, kwargs, expected_message
Review Comment:
```suggestion
self, _get_dttm_filter, get_count, soft_fail, expected_exception, kwargs, expected_failed_message
```
##########
tests/sensors/test_external_task_sensor.py:
##########
@@ -838,6 +839,172 @@ def test_external_task_group_when_there_is_no_TIs(self):
ignore_ti_state=True,
)
+ @pytest.mark.parametrize(
+ "soft_fail, expected_exception, kwargs, expected_message",
+ (
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"Some of the external tasks {re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}"
+ f" in DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"The external task_group '{re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}'"
+ f" in DAG '{TEST_DAG_ID}' failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {"failed_states": [State.FAILED]},
+ f"The external DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {"failed_states": [State.FAILED]},
+ "Skipping due to soft_fail is set to True.",
+ ),
+ ),
+ )
+ @mock.patch("airflow.sensors.external_task.ExternalTaskSensor.get_count")
+ @mock.patch("airflow.sensors.external_task.ExternalTaskSensor._get_dttm_filter")
+ def test_fail_poke(
+ self, _get_dttm_filter, get_count, soft_fail, expected_exception, kwargs, expected_message
+ ):
+ _get_dttm_filter.return_value = []
+ get_count.return_value = 1
+ op = ExternalTaskSensor(
+ task_id="test_external_task_duplicate_task_ids",
+ external_dag_id=TEST_DAG_ID,
+ allowed_states=["success"],
+ dag=self.dag,
+ soft_fail=soft_fail,
+ deferrable=False,
+ **kwargs,
+ )
Review Comment:
expec
```suggestion
)
expected_message = "Skipping due to soft_fail is set to True." if soft_fail else expected_failed_message
```
##########
tests/sensors/test_external_task_sensor.py:
##########
@@ -838,6 +839,172 @@ def test_external_task_group_when_there_is_no_TIs(self):
ignore_ti_state=True,
)
+ @pytest.mark.parametrize(
+ "soft_fail, expected_exception, kwargs, expected_message",
+ (
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"Some of the external tasks {re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}"
+ f" in DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"The external task_group '{re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}'"
+ f" in DAG '{TEST_DAG_ID}' failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {"failed_states": [State.FAILED]},
+ f"The external DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {"failed_states": [State.FAILED]},
+ "Skipping due to soft_fail is set to True.",
+ ),
+ ),
+ )
+ @mock.patch("airflow.sensors.external_task.ExternalTaskSensor.get_count")
+ @mock.patch("airflow.sensors.external_task.ExternalTaskSensor._get_dttm_filter")
+ def test_fail_poke(
+ self, _get_dttm_filter, get_count, soft_fail, expected_exception, kwargs, expected_message
+ ):
+ _get_dttm_filter.return_value = []
+ get_count.return_value = 1
+ op = ExternalTaskSensor(
+ task_id="test_external_task_duplicate_task_ids",
+ external_dag_id=TEST_DAG_ID,
+ allowed_states=["success"],
+ dag=self.dag,
+ soft_fail=soft_fail,
+ deferrable=False,
+ **kwargs,
+ )
+ with pytest.raises(expected_exception, match=expected_message):
+ op.execute(context={})
+
+ @pytest.mark.parametrize(
Review Comment:
I might try something like suggested above.
##########
tests/sensors/test_external_task_sensor.py:
##########
@@ -838,6 +839,172 @@ def test_external_task_group_when_there_is_no_TIs(self):
ignore_ti_state=True,
)
+ @pytest.mark.parametrize(
+ "soft_fail, expected_exception, kwargs, expected_message",
+ (
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"Some of the external tasks {re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}"
+ f" in DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ f"The external task_group '{re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}'"
+ f" in DAG '{TEST_DAG_ID}' failed.",
+ ),
+ (
+ False,
+ AirflowException,
+ {"failed_states": [State.FAILED]},
+ f"The external DAG {TEST_DAG_ID} failed.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {
+ "external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
+ "failed_states": [State.FAILED],
+ },
+ "Skipping due to soft_fail is set to True.",
+ ),
+ (
+ True,
+ AirflowSkipException,
+ {"failed_states": [State.FAILED]},
+ "Skipping due to soft_fail is set to True.",
+ ),
+ ),
+ )
Review Comment:
```suggestion
@pytest.mark.parametrize(
"soft_fail, expected_exception",
(
(False, AirflowException),
(True, AirflowSkipException)
)
)
@pytest.mark.parametrize(
"soft_fail, expected_exception, kwargs, expected_failed_message",
(
(
{
"external_task_ids": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
"failed_states": [State.FAILED],
},
f"Some of the external tasks {re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}"
f" in DAG {TEST_DAG_ID} failed.",
),
(
{
"external_task_group_id": [TEST_TASK_ID, TEST_TASK_ID_ALTERNATE],
"failed_states": [State.FAILED],
},
f"The external task_group '{re.escape(str([TEST_TASK_ID, TEST_TASK_ID_ALTERNATE]))}'"
f" in DAG '{TEST_DAG_ID}' failed.",
),
(
{"failed_states": [State.FAILED]},
f"The external DAG {TEST_DAG_ID} failed.",
),
),
)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org