Add test for interval timetable catchup=False

[uranusjr]

Test case for #19130 + some minor fixes I missed earlier.

cc @robinedwards

[uranusjr]

The previous implementation in #19130 missed this call. Without this alignment call, the next run's data interval would start at the current time (e.g. 2021-10-22T10:43:57.43256) instead of the interval boundary (e.g. 2021-10-22T10:00:00.00000) and cause inconsistencies. So I moved the call here to make sure it is always done correctly.

[robinedwards]

Brilliant good spot

[uranusjr]

The previous implementation

next_start = self._get_next(current_time)
last_start = self._get_prev(current_time)

had a bug if current_time falls right on the interval boundary (e.g. the full hour mark for a @hourly schedule interval) because croniter would make next_start and last_start two hours apart instead of one. So this is changed to

last_start = self._get_prev(current_time)
next_start = self._get_next(last_start)

toe ensure the two starts are one hour apart, and current_time == next_start for the interval boundary case. This is not really practically relevant (what's the chance a now() call falls directly one that time with microsecond accurary), but is an issue in unit tests.

[uranusjr]

tests/models/test_cleartasks.py::TestClearTasks::test_clear_task_instances: assert 3 == 1
 +  where 3 = <TaskInstance: test_clear_task_instances.1 scheduled__2016-01-01T00:00:00+00:00 [None]>.max_tries
tests/models/test_cleartasks.py::TestClearTasks::test_clear_task_instances_with_task_reschedule: AssertionError: assert 0 == 1
 +  where 0 = <function TestClearTasks.test_clear_task_instances_with_task_reschedule.<locals>.count_task_reschedule at 0x7f3f4c267b70>('1')
 +    where '1' = <TaskInstance: test_clear_task_instances_with_task_reschedule.1 scheduled__2016-01-01T00:00:00+00:00 [up_for_reschedule]>.task_id
tests/models/test_cleartasks.py::TestClearTasks::test_dag_clear: assert 3 == 1
 +  where 3 = <TaskInstance: test_dag_clear.test_dag_clear_task_1 scheduled__2016-01-01T00:00:00+00:00 [None]>.max_tries

Only failing on Postgres, so probably not related...? But DAG scheduling is so core I can't really be sure.

[ephraimbuddy]

Can we also add the test I dropped at #19130, might be useful

[uranusjr]

Added (with some minor changes, adding session=session etc.)

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[uranusjr]

Test failures went away, probably flaky.

[robinedwards]

Perhaps worth a comment here as to why we are doing this. Or you could freezetime to queue and schedule dagruns at the dag start date then outside of the freezetime block queue another dagrun and then confirm its the latest execution date?