ashb commented on a change in pull request #18061:
URL: https://github.com/apache/airflow/pull/18061#discussion_r704184367
##########
File path: tests/jobs/test_scheduler_job.py
##########
@@ -2689,6 +2689,101 @@ def
test_max_active_runs_in_a_dag_doesnt_stop_running_dagruns_in_otherdags(self,
)
assert len(session.query(DagRun).filter(DagRun.state ==
State.RUNNING).all()) == 11
+ def test_start_queued_dagruns_do_follow_execution_date_order(self,
dag_maker):
+ session = settings.Session()
+ with dag_maker('test_dag1', max_active_runs=1) as dag:
+ DummyOperator(task_id='mytask')
+ date = dag.following_schedule(DEFAULT_DATE)
+ for i in range(30):
+ dr = dag_maker.create_dagrun(
+ run_id=f'dagrun_{i}', run_type=DagRunType.SCHEDULED,
state=State.QUEUED, execution_date=date
+ )
+ date = dr.execution_date + timedelta(hours=1)
+ self.scheduler_job = SchedulerJob(subdir=os.devnull)
+ self.scheduler_job.executor = MockExecutor(do_update=False)
+ self.scheduler_job.processor_agent =
mock.MagicMock(spec=DagFileProcessorAgent)
+
+ self.scheduler_job._start_queued_dagruns(session)
+ session.flush()
+ dr = DagRun.find(run_id='dagrun_0')
+ ti = dr[0].get_task_instance(task_id='mytask', session=session)
+ ti.state = State.SUCCESS
+ session.merge(ti)
+ session.commit()
+ assert dr[0].state == State.RUNNING
+ dr[0].state = State.SUCCESS
+ session.merge(dr[0])
+ session.flush()
+ assert dr[0].state == State.SUCCESS
+ self.scheduler_job._start_queued_dagruns(session)
+ session.flush()
+ dr = DagRun.find(run_id='dagrun_1')
+ assert len(session.query(DagRun).filter(DagRun.state ==
State.RUNNING).all()) == 1
+
+ assert dr[0].state == State.RUNNING
+
+ def test_no_dagruns_would_stuck_in_running(self, dag_maker):
+ # Test that running dagruns are not stuck in running.
+ # Create one dagrun in 'running' state and 1 in 'queued' state from
one dag(max_active_runs=1)
+ # Create 16 dagruns in 'running' state and 16 in 'queued' state from
another dag
+ # Create 16 dagruns in 'running' state and 16 in 'queued' state from
yet another dag
+ # Finish the task of the first dag, and check that another dagrun
starts running
+ # from the first dag.
+
+ session = settings.Session()
+ # first dag and dagruns
+ with dag_maker('test_dagrun_states_are_correct_1', max_active_runs=1)
as dag:
+ task1 = DummyOperator(task_id='dummy_task')
+ date = DEFAULT_DATE
Review comment:
Please have this be explicit:
```python
date = timezone.datetime(2016, 1, 1)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
