This is an automated email from the ASF dual-hosted git repository.
vincbeck 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 ff9576347ee Fix region_name being ignored by the Step Functions
execution trigger (#72625)
ff9576347ee is described below
commit ff9576347eee591f493540c79162a3caf7833660
Author: Nandesh Kanagaraju <[email protected]>
AuthorDate: Wed Sep 9 21:38:01 2026 +0530
Fix region_name being ignored by the Step Functions execution trigger
(#72625)
StepFunctionsExecutionCompleteTrigger accepted region_name but only put
it into serialized_fields without forwarding it to AwsBaseWaiterTrigger,
so self.region_name stayed None and the deferred waiter always polled
Step Functions in the default region instead of the one the operator was
configured with. The base class already serializes region_name itself,
so the serialized_fields entry was redundant on top of being inert.
Also adds the previously missing dedicated test module for the trigger
(tracked in the OVERLOOKED_TESTS allowlist), including a regression test
that fails without this fix.
---
.../tests/unit/always/test_project_structure.py | 1 -
.../providers/amazon/aws/triggers/step_function.py | 3 +-
.../unit/amazon/aws/triggers/test_step_function.py | 89 ++++++++++++++++++++++
3 files changed, 91 insertions(+), 2 deletions(-)
diff --git a/airflow-core/tests/unit/always/test_project_structure.py
b/airflow-core/tests/unit/always/test_project_structure.py
index 2efbc1d7470..f59a1f77823 100644
--- a/airflow-core/tests/unit/always/test_project_structure.py
+++ b/airflow-core/tests/unit/always/test_project_structure.py
@@ -74,7 +74,6 @@ class TestProjectStructure:
"providers/amazon/tests/unit/amazon/aws/sensors/test_sagemaker.py",
"providers/amazon/tests/unit/amazon/aws/test_exceptions.py",
"providers/amazon/tests/unit/amazon/aws/triggers/test_sagemaker_unified_studio.py",
-
"providers/amazon/tests/unit/amazon/aws/triggers/test_step_function.py",
"providers/amazon/tests/unit/amazon/aws/utils/test_rds.py",
"providers/amazon/tests/unit/amazon/aws/utils/test_sagemaker.py",
"providers/amazon/tests/unit/amazon/aws/waiters/test_base_waiter.py",
diff --git
a/providers/amazon/src/airflow/providers/amazon/aws/triggers/step_function.py
b/providers/amazon/src/airflow/providers/amazon/aws/triggers/step_function.py
index 6fe6af22184..a52f1909696 100644
---
a/providers/amazon/src/airflow/providers/amazon/aws/triggers/step_function.py
+++
b/providers/amazon/src/airflow/providers/amazon/aws/triggers/step_function.py
@@ -46,7 +46,7 @@ class
StepFunctionsExecutionCompleteTrigger(AwsBaseWaiterTrigger):
**kwargs,
) -> None:
super().__init__(
- serialized_fields={"execution_arn": execution_arn, "region_name":
region_name},
+ serialized_fields={"execution_arn": execution_arn},
waiter_name="step_function_succeeded",
waiter_args={"executionArn": execution_arn},
failure_message="Step function failed",
@@ -57,6 +57,7 @@ class
StepFunctionsExecutionCompleteTrigger(AwsBaseWaiterTrigger):
waiter_delay=waiter_delay,
waiter_max_attempts=waiter_max_attempts,
aws_conn_id=aws_conn_id,
+ region_name=region_name,
**kwargs,
)
diff --git
a/providers/amazon/tests/unit/amazon/aws/triggers/test_step_function.py
b/providers/amazon/tests/unit/amazon/aws/triggers/test_step_function.py
new file mode 100644
index 00000000000..3341952450e
--- /dev/null
+++ b/providers/amazon/tests/unit/amazon/aws/triggers/test_step_function.py
@@ -0,0 +1,89 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+from __future__ import annotations
+
+from unittest import mock
+from unittest.mock import AsyncMock
+
+import pytest
+
+from airflow.providers.amazon.aws.hooks.step_function import StepFunctionHook
+from airflow.providers.amazon.aws.triggers.step_function import
StepFunctionsExecutionCompleteTrigger
+from airflow.triggers.base import TriggerEvent
+
+BASE_TRIGGER_CLASSPATH = "airflow.providers.amazon.aws.triggers.step_function."
+
+
+class TestStepFunctionsExecutionCompleteTrigger:
+ EXPECTED_WAITER_NAME = "step_function_succeeded"
+ EXECUTION_ARN = (
+ "arn:aws:states:us-east-1:123456789012:execution:"
+ "pseudo-state-machine:020f5b16-b1a1-4149-946f-92dd32d97934"
+ )
+
+ def test_serialization(self):
+ trigger = StepFunctionsExecutionCompleteTrigger(
+ execution_arn=self.EXECUTION_ARN,
+ waiter_delay=10,
+ waiter_max_attempts=5,
+ aws_conn_id="aws_step_function_conn",
+ region_name="eu-central-1",
+ )
+
+ classpath, kwargs = trigger.serialize()
+
+ assert classpath == BASE_TRIGGER_CLASSPATH +
"StepFunctionsExecutionCompleteTrigger"
+ assert kwargs.get("execution_arn") == self.EXECUTION_ARN
+ assert kwargs.get("waiter_delay") == 10
+ assert kwargs.get("waiter_max_attempts") == 5
+ assert kwargs.get("aws_conn_id") == "aws_step_function_conn"
+ assert kwargs.get("region_name") == "eu-central-1"
+
+ def test_hook_forwards_connection_config(self):
+ trigger = StepFunctionsExecutionCompleteTrigger(
+ execution_arn=self.EXECUTION_ARN,
+ aws_conn_id="aws_step_function_conn",
+ region_name="eu-central-1",
+ verify=False,
+ botocore_config={"read_timeout": 100},
+ )
+
+ hook = trigger.hook()
+
+ assert isinstance(hook, StepFunctionHook)
+ assert hook.aws_conn_id == "aws_step_function_conn"
+ assert hook._region_name == "eu-central-1"
+ assert hook._verify is False
+ assert hook._config.read_timeout == 100
+
+ @pytest.mark.asyncio
+ @mock.patch.object(StepFunctionHook, "get_waiter")
+ @mock.patch.object(StepFunctionHook, "get_async_conn")
+ async def test_run_success(self, mock_async_conn, mock_get_waiter):
+ mock_async_conn.return_value.__aenter__.return_value = mock.MagicMock()
+ mock_get_waiter().wait = AsyncMock()
+ trigger =
StepFunctionsExecutionCompleteTrigger(execution_arn=self.EXECUTION_ARN)
+
+ generator = trigger.run()
+ response = await generator.asend(None)
+
+ assert response == TriggerEvent({"status": "success", "execution_arn":
self.EXECUTION_ARN})
+ assert mock_get_waiter().wait.call_count == 1
+ mock_get_waiter.assert_any_call(
+ self.EXPECTED_WAITER_NAME, deferrable=True, client=mock.ANY,
config_overrides=None
+ )
+ assert mock_get_waiter().wait.call_args.kwargs["executionArn"] ==
self.EXECUTION_ARN
