Ensure execution_timeout as timedelta (#23655)

airflow/models/baseoperator.py

@@ -767,7 +767,13 @@ def __init__(
         self.email = email
         self.email_on_retry = email_on_retry
         self.email_on_failure = email_on_failure
+
+        if execution_timeout is not None and not isinstance(execution_timeout, timedelta):
+            raise ValueError(
+                f'execution_timeout must be timedelta object but passed as type: {type(execution_timeout)}'
+            )
         self.execution_timeout = execution_timeout
+
         self.on_execute_callback = on_execute_callback
         self.on_failure_callback = on_failure_callback
         self.on_success_callback = on_success_callback

tests/models/test_baseoperator.py

@@ -119,6 +119,17 @@ def test_default_args(self):
         with pytest.raises(AirflowException, match="missing keyword argument 'test_sub_param'"):
             DummySubClass(default_args=default_args)
 
+    def test_execution_timeout_type(self):
+        with pytest.raises(
+            ValueError, match="execution_timeout must be timedelta object but passed as type: <class 'str'>"
+        ):
+            BaseOperator(task_id='test', execution_timeout="1")
+
+        with pytest.raises(
+            ValueError, match="execution_timeout must be timedelta object but passed as type: <class 'int'>"
+        ):
+            BaseOperator(task_id='test', execution_timeout=1)
+
     def test_incorrect_default_args(self):
         default_args = {'test_param': True, 'extra_param': True}
         dummy_class = DummyClass(default_args=default_args)