This is an automated email from the ASF dual-hosted git repository.
jscheffl 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 5fc030aeac3 Add tests for execute_complete success and error paths
(#64119)
5fc030aeac3 is described below
commit 5fc030aeac34ea518b65cdfe84bb0fad17d2704b
Author: SameerMesiah97 <[email protected]>
AuthorDate: Mon Apr 20 19:23:18 2026 +0100
Add tests for execute_complete success and error paths (#64119)
Covers behavior where execute_complete returns successfully on non-error
events and raises on error events. AirflowException has been replaced with
RuntimeError.
Co-authored-by: Sameer Mesiah <[email protected]>
---
.../airflow/providers/airbyte/operators/airbyte.py | 2 +-
.../tests/unit/airbyte/operators/test_airbyte.py | 47 ++++++++++++++++++++++
scripts/ci/prek/known_airflow_exceptions.txt | 2 +-
3 files changed, 49 insertions(+), 2 deletions(-)
diff --git
a/providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py
b/providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py
index 9900e7adf17..b3537504cf1 100644
--- a/providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py
+++ b/providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py
@@ -127,7 +127,7 @@ class AirbyteTriggerSyncOperator(BaseOperator):
"""
if event["status"] == "error":
self.log.debug("Error occurred with context: %s", context)
- raise AirflowException(event["message"])
+ raise RuntimeError(event["message"])
self.log.info("%s completed successfully.", self.task_id)
return None
diff --git a/providers/airbyte/tests/unit/airbyte/operators/test_airbyte.py
b/providers/airbyte/tests/unit/airbyte/operators/test_airbyte.py
index 951194a546a..7746ca35cf8 100644
--- a/providers/airbyte/tests/unit/airbyte/operators/test_airbyte.py
+++ b/providers/airbyte/tests/unit/airbyte/operators/test_airbyte.py
@@ -19,6 +19,7 @@ from __future__ import annotations
from unittest import mock
+import pytest
from airbyte_api.models import JobCreateRequest, JobResponse, JobStatusEnum,
JobTypeEnum
from airflow.models import Connection
@@ -67,6 +68,52 @@ class TestAirbyteTriggerSyncOp:
job_id=self.job_id, wait_seconds=self.wait_seconds,
timeout=self.timeout
)
+ @pytest.mark.parametrize("status", ["success", "cancelled"])
+ def test_execute_complete_non_error_states(self, status,
create_connection_without_db):
+ conn = Connection(conn_id=self.airbyte_conn_id, conn_type="airbyte",
host="airbyte.com")
+ create_connection_without_db(conn)
+
+ op = AirbyteTriggerSyncOperator(
+ task_id="test_airbyte_op",
+ airbyte_conn_id=self.airbyte_conn_id,
+ connection_id=self.connection_id,
+ wait_seconds=self.wait_seconds,
+ timeout=self.timeout,
+ deferrable=True,
+ )
+
+ event = {
+ "status": status,
+ "message": "succeeded/cancelled",
+ "job_id": self.job_id,
+ }
+
+ result = op.execute_complete(context={}, event=event)
+
+ assert result is None
+
+ def test_execute_complete_error(self, create_connection_without_db):
+ conn = Connection(conn_id=self.airbyte_conn_id, conn_type="airbyte",
host="airbyte.com")
+ create_connection_without_db(conn)
+
+ op = AirbyteTriggerSyncOperator(
+ task_id="test_airbyte_op",
+ airbyte_conn_id=self.airbyte_conn_id,
+ connection_id=self.connection_id,
+ wait_seconds=self.wait_seconds,
+ timeout=self.timeout,
+ deferrable=True,
+ )
+
+ error_event = {
+ "status": "error",
+ "message": "Job failed",
+ "job_id": self.job_id,
+ }
+
+ with pytest.raises(RuntimeError, match="Job failed"):
+ op.execute_complete(context={}, event=error_event)
+
@mock.patch("airflow.providers.airbyte.hooks.airbyte.AirbyteHook.get_job_status")
@mock.patch("airflow.providers.airbyte.hooks.airbyte.AirbyteHook.cancel_job")
def test_on_kill(self, mock_cancel_job, mock_get_job_status,
create_connection_without_db):
diff --git a/scripts/ci/prek/known_airflow_exceptions.txt
b/scripts/ci/prek/known_airflow_exceptions.txt
index 1a870c169d6..c44440cb5a3 100644
--- a/scripts/ci/prek/known_airflow_exceptions.txt
+++ b/scripts/ci/prek/known_airflow_exceptions.txt
@@ -32,7 +32,7 @@
devel-common/src/tests_common/test_utils/salesforce_system_helpers.py::2
devel-common/src/tests_common/test_utils/sftp_system_helpers.py::1
devel-common/src/tests_common/test_utils/watcher.py::1
providers/airbyte/src/airflow/providers/airbyte/hooks/airbyte.py::8
-providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py::4
+providers/airbyte/src/airflow/providers/airbyte/operators/airbyte.py::3
providers/airbyte/src/airflow/providers/airbyte/sensors/airbyte.py::6
providers/alibaba/src/airflow/providers/alibaba/cloud/hooks/analyticdb_spark.py::6
providers/alibaba/src/airflow/providers/alibaba/cloud/hooks/oss.py::11
