kaxil commented on code in PR #65207:
URL: https://github.com/apache/airflow/pull/65207#discussion_r3097484027
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -829,6 +829,161 @@ def add_failed_dag_run(dag, id, logical_date):
add_failed_dag_run(scheduler_dag, "2", TEST_DATE + timedelta(days=1))
assert session.get(DagModel, dag.dag_id).is_paused
+ def test_dag_paused_after_limit_orders_by_run_after(self,
testing_dag_bundle):
Review Comment:
These two new tests repeat ~65 lines of nearly identical `create_dagrun` +
`set_state` + `update_state` setup. The neighbor
`test_existing_dag_is_paused_after_limit` above uses a local
`add_failed_dag_run` helper for a reason. Consider extracting a helper that
takes `(run_id, logical_date, run_after, ti_state, run_state)` so a future
`create_dagrun` signature change does not require touching six call sites.
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -829,6 +829,161 @@ def add_failed_dag_run(dag, id, logical_date):
add_failed_dag_run(scheduler_dag, "2", TEST_DATE + timedelta(days=1))
assert session.get(DagModel, dag.dag_id).is_paused
+ def test_dag_paused_after_limit_orders_by_run_after(self,
testing_dag_bundle):
+ """Verify _check_last_n_dagruns_failed orders by run_after, not
logical_date.
+
+ Create three runs where run_after ordering differs from logical_date
ordering.
+ The two most-recent runs by run_after are FAILED, while the oldest
(which has the
+ latest logical_date) is SUCCESS. The DAG should be paused because the
check
+ must use run_after ordering.
+ """
+ dag_id = "dag_paused_orders_by_run_after"
+ dag = DAG(dag_id, schedule=None, is_paused_upon_creation=False,
max_consecutive_failed_dag_runs=2)
+ op1 = BashOperator(task_id="task", bash_command="exit 1;")
+ dag.add_task(op1)
+ session = settings.Session()
+ orm_dag = DagModel(
+ dag_id=dag.dag_id,
+ bundle_name="testing",
+ is_stale=False,
+ )
+ session.add(orm_dag)
+ session.flush()
+
+ scheduler_dag = sync_dag_to_db(dag, session=session)
+ assert not session.get(DagModel, dag.dag_id).is_paused
+
+ # Run 1: oldest by run_after but LATEST logical_date — SUCCESS
+ dr1 = scheduler_dag.create_dagrun(
+ run_type=DagRunType.MANUAL,
+ run_id="run_success",
+ logical_date=TEST_DATE + timedelta(days=10),
+ state=State.SUCCESS,
+ data_interval=(TEST_DATE, TEST_DATE),
+ run_after=TEST_DATE,
+ triggered_by=DagRunTriggeredByType.TEST,
+ session=session,
+ )
+ ti1 = dr1.get_task_instance(task_id=op1.task_id, session=session)
+ ti1.set_state(state=TaskInstanceState.SUCCESS, session=session)
+ dr1.update_state(session=session)
Review Comment:
`dr1` is SUCCESS here, and `DagRun.update_state` only calls
`_check_last_n_dagruns_failed` on the FAILED branch, so this call is a no-op
for what the test is exercising. Fine to keep if you want to mirror the real
flow, but a short comment saying so would help, otherwise it is safe to drop.
--
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]
