This is an automated email from the ASF dual-hosted git repository. ephraimanierobi pushed a commit to branch v2-4-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit dc3817cc2c38e59d445d97d68350cc37775ba483 Author: paolomoriello <[email protected]> AuthorDate: Sun Nov 6 18:07:26 2022 +0100 Move TriggerDagRun conf check to execute (#27035) (cherry picked from commit 2462d57eccd8353b9a9d32a91d75c027e137fd7c) --- airflow/operators/trigger_dagrun.py | 10 +++++----- tests/operators/test_trigger_dagrun.py | 14 +++++++------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/airflow/operators/trigger_dagrun.py b/airflow/operators/trigger_dagrun.py index 630ca9c354..c758f90ce0 100644 --- a/airflow/operators/trigger_dagrun.py +++ b/airflow/operators/trigger_dagrun.py @@ -115,11 +115,6 @@ class TriggerDagRunOperator(BaseOperator): self.execution_date = execution_date - try: - json.dumps(self.conf) - except TypeError: - raise AirflowException("conf parameter should be JSON Serializable") - def execute(self, context: Context): if isinstance(self.execution_date, datetime.datetime): parsed_execution_date = self.execution_date @@ -128,6 +123,11 @@ class TriggerDagRunOperator(BaseOperator): else: parsed_execution_date = timezone.utcnow() + try: + json.dumps(self.conf) + except TypeError: + raise AirflowException("conf parameter should be JSON Serializable") + if self.trigger_run_id: run_id = self.trigger_run_id else: diff --git a/tests/operators/test_trigger_dagrun.py b/tests/operators/test_trigger_dagrun.py index d9b88fedb7..e391273069 100644 --- a/tests/operators/test_trigger_dagrun.py +++ b/tests/operators/test_trigger_dagrun.py @@ -215,14 +215,14 @@ class TestDagRunOperator(TestCase): def test_trigger_dagrun_operator_templated_invalid_conf(self): """Test passing a conf that is not JSON Serializable raise error.""" - + task = TriggerDagRunOperator( + task_id="test_trigger_dagrun_with_invalid_conf", + trigger_dag_id=TRIGGERED_DAG_ID, + conf={"foo": "{{ dag.dag_id }}", "datetime": timezone.utcnow()}, + dag=self.dag, + ) with pytest.raises(AirflowException, match="^conf parameter should be JSON Serializable$"): - TriggerDagRunOperator( - task_id="test_trigger_dagrun_with_invalid_conf", - trigger_dag_id=TRIGGERED_DAG_ID, - conf={"foo": "{{ dag.dag_id }}", "datetime": timezone.utcnow()}, - dag=self.dag, - ) + task.run(start_date=DEFAULT_DATE, end_date=DEFAULT_DATE) def test_trigger_dagrun_operator_templated_conf(self): """Test passing a templated conf to the triggered DagRun."""
