Add 'reschedule' to the serialized fields for the BaseSensorOperator (#…

…23674)

fix #23411

(cherry picked from commit f9e2a30)

airflow/sensors/base.py

@@ -339,6 +339,10 @@ def reschedule(self):
         """Define mode rescheduled sensors."""
         return self.mode == 'reschedule'
 
+    @classmethod
+    def get_serialized_fields(cls):
+        return super().get_serialized_fields() | {"reschedule"}
+
 
 def poke_mode_only(cls):
     """

tests/serialization/test_dag_serialization.py

@@ -1462,6 +1462,7 @@ def poke(self, context: Context):
         assert "deps" in blob
 
         serialized_op = SerializedBaseOperator.deserialize_operator(blob)
+        assert serialized_op.reschedule == (mode == "reschedule")
         assert op.deps == serialized_op.deps
 
     @pytest.mark.parametrize(

tests/ti_deps/deps/test_ready_to_reschedule_dep.py

@@ -31,7 +31,7 @@
 class TestNotInReschedulePeriodDep(unittest.TestCase):
     def _get_task_instance(self, state):
         dag = DAG('test_dag')
-        task = Mock(dag=dag)
+        task = Mock(dag=dag, reschedule=True)
         ti = TaskInstance(task=task, state=state, run_id=None)
         return ti
 
@@ -52,6 +52,11 @@ def test_should_pass_if_ignore_in_reschedule_period_is_set(self):
         dep_context = DepContext(ignore_in_reschedule_period=True)
         assert ReadyToRescheduleDep().is_met(ti=ti, dep_context=dep_context)
 
+    def test_should_pass_if_not_reschedule_mode(self):
+        ti = self._get_task_instance(State.UP_FOR_RESCHEDULE)
+        del ti.task.reschedule
+        assert ReadyToRescheduleDep().is_met(ti=ti)
+
     def test_should_pass_if_not_in_none_state(self):
         ti = self._get_task_instance(State.UP_FOR_RETRY)
         assert ReadyToRescheduleDep().is_met(ti=ti)