From 29d02c56f5cada4896ae00613ab15574b5d29746 Mon Sep 17 00:00:00 2001 From: Kirilishin Aleksei Date: Sat, 9 Dec 2023 03:24:19 +0400 Subject: [PATCH] Add the empty timetable tests --- airflow/timetables/events.py | 13 ++++++++----- tests/timetables/test_events_timetable.py | 15 +++++++++++++++ 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/airflow/timetables/events.py b/airflow/timetables/events.py index 53d5019da4226..0edf1b101299c 100644 --- a/airflow/timetables/events.py +++ b/airflow/timetables/events.py @@ -58,10 +58,13 @@ def __init__( self.event_dates.sort() self.restrict_to_events = restrict_to_events if description is None: - self.description = ( - f"{len(self.event_dates)} Events between {self.event_dates[0]} and {self.event_dates[-1]}" - ) - self._summary = f"{len(self.event_dates)} Events" + if self.event_dates: + self.description = ( + f"{len(self.event_dates)} events between {self.event_dates[0]} and {self.event_dates[-1]}" + ) + else: + self.description = "No events" + self._summary = f"{len(self.event_dates)} events" else: self._summary = description self.description = description @@ -106,7 +109,7 @@ def next_dagrun_info( def infer_manual_data_interval(self, *, run_after: DateTime) -> DataInterval: # If Timetable not restricted to events, run for the time specified - if not self.restrict_to_events: + if not self.restrict_to_events or not self.event_dates: return DataInterval.exact(run_after) # If restricted to events, run for the most recent past event diff --git a/tests/timetables/test_events_timetable.py b/tests/timetables/test_events_timetable.py index 131a618de6b8d..39d1fd3431e34 100644 --- a/tests/timetables/test_events_timetable.py +++ b/tests/timetables/test_events_timetable.py @@ -161,3 +161,18 @@ def test_no_catchup_first_starts( restriction=TimeRestriction(earliest=START_DATE, latest=None, catchup=False), ) assert next_info == expected_info + + +def test_empty_timetable() -> None: + empty_timetable = EventsTimetable(event_dates=[]) + next_info = empty_timetable.next_dagrun_info( + last_automated_data_interval=None, + restriction=TimeRestriction(earliest=START_DATE, latest=None, catchup=False), + ) + assert next_info is None + + +def test_empty_timetable_manual_run() -> None: + empty_timetable = EventsTimetable(event_dates=[]) + manual_run_data_interval = empty_timetable.infer_manual_data_interval(run_after=START_DATE) + assert manual_run_data_interval == DataInterval(start=START_DATE, end=START_DATE)