luc-pimentel commented on code in PR #75170:
URL: https://github.com/apache/airflow/pull/75170#discussion_r4407618059
##########
airflow-core/src/airflow/api_fastapi/core_api/services/public/task_instances.py:
##########
@@ -668,6 +709,22 @@ def _categorize_entities(
)
continue
+ if isinstance(entity, BulkTaskInstanceBody) and
(entity.include_past or entity.include_future):
+ run_key = (dag_id, dag_run_id)
+ if run_key not in dag_run_cache:
+ dag_run_cache[run_key] = self.session.scalar(
+ select(DagRun).where(DagRun.dag_id == dag_id,
DagRun.run_id == dag_run_id)
+ )
+ dag_run = dag_run_cache[run_key]
+ if dag_run is not None:
+ try:
+ validate_past_future_for_logical_date(
+ entity.include_past, entity.include_future,
dag_run.logical_date
+ )
+ except ValueError as err:
+ results.errors.append({"error": str(err),
"status_code": status.HTTP_400_BAD_REQUEST})
Review Comment:
`_categorize_entities` returns the no-logical-date error before checking
access to the Dag. Should we move this validation below the authorization check?
With a user outside the Dag team, I get a per-item 400 for an existing
dateless run but 403 for dated or missing runs. On the base revision, all three
return 403. This exposes run information to a caller who cannot access that Dag.
---
Drafted-by: OpenCode 1.19.0 (GPT-6 Astra); reviewed by @luc-pimentel before
posting
##########
airflow-core/tests/unit/models/test_dag.py:
##########
@@ -3739,6 +3739,49 @@ def get_ti_from_db(task):
}
[email protected]("flag", ["future", "past"])
+def test_set_task_instance_state_no_logical_date_raises_value_error(session,
dag_maker, flag):
+ """Test that set_task_instance_state with past/future raises ValueError on
a run with no logical_date."""
+ from airflow.serialization.definitions.dag import
NO_LOGICAL_DATE_PAST_FUTURE_ERROR
+
+ with dag_maker("test_set_ti_no_logical_date", session=session,
serialized=True) as dag:
+ EmptyOperator(task_id="task_1") >> EmptyOperator(task_id="task_2")
+
+ dagrun = dag_maker.create_dagrun(run_id="manual",
run_type=DagRunType.MANUAL, logical_date=None)
+
+ with pytest.raises(ValueError,
match=re.escape(NO_LOGICAL_DATE_PAST_FUTURE_ERROR)):
+ dag.set_task_instance_state(
+ task_id="task_1",
+ run_id=dagrun.run_id,
+ state=State.SUCCESS,
+ session=session,
+ **{flag: True},
+ )
Review Comment:
This test checks the exception but not whether the task changed first.
Should we snapshot the task states before the call, then compare them after
flushing and reloading the session?
before = {ti.task_id: ti.state for ti in
dagrun.get_task_instances(session=session)}
# Existing call and ValueError assertion go here.
session.flush()
session.expire_all()
assert {ti.task_id: ti.state for ti in
dagrun.get_task_instances(session=session)} == before
Moving the guard below `set_state` leaves all 18 new tests passing, although
task_1 becomes success. The state-preservation assertion fails for both flags
on that mutation and passes for both on this head.
---
Drafted-by: OpenCode 1.19.0 (GPT-6 Astra); reviewed by @luc-pimentel before
posting
--
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]
