diff --git a/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py b/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py index aab675201497c..0209e5b737bd0 100644 --- a/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py +++ b/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py @@ -140,8 +140,6 @@ def __init__( ), **kwargs, ): - if not exactly_one(job_flow_id is None, job_flow_name is None): - raise AirflowException("Exactly one of job_flow_id or job_flow_name must be specified.") super().__init__(**kwargs) cluster_states = cluster_states or [] steps = steps or [] @@ -186,6 +184,9 @@ def _inject_openlineage_parent_job_information(self, steps: list[dict], context: return result def execute(self, context: Context) -> list[str]: + if not exactly_one(self.job_flow_id is None, self.job_flow_name is None): + raise AirflowException("Exactly one of job_flow_id or job_flow_name must be specified.") + job_flow_id = self.job_flow_id or self.hook.get_cluster_id_by_name( str(self.job_flow_name), self.cluster_states ) diff --git a/providers/amazon/tests/unit/amazon/aws/operators/test_emr_add_steps.py b/providers/amazon/tests/unit/amazon/aws/operators/test_emr_add_steps.py index 7a8d6b3892ed9..1e00389c93466 100644 --- a/providers/amazon/tests/unit/amazon/aws/operators/test_emr_add_steps.py +++ b/providers/amazon/tests/unit/amazon/aws/operators/test_emr_add_steps.py @@ -91,14 +91,15 @@ def test_init(self): pytest.param(None, None, id="both-none"), ], ) - def test_validate_mutually_exclusive_args(self, job_flow_id, job_flow_name): + def test_validate_mutually_exclusive_args_on_execute(self, job_flow_id, job_flow_name): + operator = EmrAddStepsOperator( + task_id="test_validate_mutually_exclusive_args", + job_flow_id=job_flow_id, + job_flow_name=job_flow_name, + ) error_message = r"Exactly one of job_flow_id or job_flow_name must be specified\." with pytest.raises(AirflowException, match=error_message): - EmrAddStepsOperator( - task_id="test_validate_mutually_exclusive_args", - job_flow_id=job_flow_id, - job_flow_name=job_flow_name, - ) + operator.execute(context=MagicMock()) @pytest.mark.db_test def test_render_template(self, session, clean_dags_dagruns_and_dagbundles, testing_dag_bundle): diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt b/scripts/ci/prek/validate_operators_init_exemptions.txt index 29ee3dd263922..cf0b4c258e93a 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -10,7 +10,6 @@ providers/amazon/src/airflow/providers/amazon/aws/operators/appflow.py::AppflowB providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py::DmsModifyTaskOperator providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py::DmsStartReplicationOperator providers/amazon/src/airflow/providers/amazon/aws/operators/ecs.py::EcsRunTaskOperator -providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py::EmrAddStepsOperator providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStartDbClusterOperator providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py::NeptuneStopDbClusterOperator providers/amazon/src/airflow/providers/amazon/aws/operators/s3.py::S3DeleteObjectsOperator