jroachgolf84 commented on code in PR #65314:
URL: https://github.com/apache/airflow/pull/65314#discussion_r3695644837
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3644,6 +3646,176 @@ def
test_clear_taskinstance_is_called_with_invalid_task_ids(self, test_client, s
assert dagrun.state == "running"
assert all(ti.state == "running" for ti in tis)
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear",
return_value=[])
+ def test_include_downstream_dags_sets_include_dependent_dags(self,
mock_clear, test_client, session):
+ """include_downstream_dags=True must be forwarded to dag.clear() as
include_dependent_dags=True."""
+ self.create_task_instances(session)
+ response = test_client.post(
+ "/dags/example_python_operator/clearTaskInstances",
+ json={"dry_run": True, "only_failed": False,
"include_downstream_dags": True},
+ )
+ assert response.status_code == 200
+ assert mock_clear.call_count == 1
+ assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear",
return_value=[])
+ def test_include_downstream_sets_include_dependent_dags(self, mock_clear,
test_client, session):
+ """include_downstream=True must also set include_dependent_dags=True
(restoring Airflow 2 behavior)."""
+ self.create_task_instances(session)
+ response = test_client.post(
+ "/dags/example_python_operator/clearTaskInstances",
+ json={"dry_run": True, "only_failed": False, "include_downstream":
True},
+ )
+ assert response.status_code == 200
+ assert mock_clear.call_count == 1
+ assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear",
return_value=[])
+ def test_default_does_not_set_include_dependent_dags(self, mock_clear,
test_client, session):
+ """Without downstream flags, include_dependent_dags must default to
False."""
+ self.create_task_instances(session)
+ response = test_client.post(
+ "/dags/example_python_operator/clearTaskInstances",
+ json={"dry_run": True, "only_failed": False},
+ )
+ assert response.status_code == 200
+ assert mock_clear.call_count == 1
+ assert mock_clear.call_args.kwargs["include_dependent_dags"] is False
+
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear",
return_value=[])
+ def test_run_id_path_sets_include_dependent_dags(self, mock_clear,
test_client, session):
+ """dag_run_id code path: include_downstream_dags=True →
include_dependent_dags=True."""
+ self.create_task_instances(session)
+ response = test_client.post(
+ "/dags/example_python_operator/clearTaskInstances",
+ json={
+ "dry_run": True,
+ "only_failed": False,
+ "dag_run_id": "TEST_DAG_RUN_ID",
+ "include_downstream_dags": True,
+ },
+ )
+ assert response.status_code == 200
+ assert mock_clear.call_count == 1
+ assert mock_clear.call_args.kwargs["include_dependent_dags"] is True
+
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear",
return_value=[])
+ def test_run_id_path_passes_request_dag_bag_to_clear(self, mock_clear,
test_client, session):
+ """dag_run_id code path: the request-scoped dag_bag must reach
dag.clear()."""
+ self.create_task_instances(session)
+ response = test_client.post(
+ "/dags/example_python_operator/clearTaskInstances",
+ json={"dry_run": True, "only_failed": False, "dag_run_id":
"TEST_DAG_RUN_ID"},
+ )
+
+ assert response.status_code == 200
+ assert mock_clear.call_count == 1
+ assert mock_clear.call_args.kwargs["dag_bag"] is
test_client.app.state.dag_bag
+
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear",
return_value=[])
+ def test_date_range_path_passes_request_dag_bag_to_clear(self, mock_clear,
test_client, session):
+ """Date-range code path (no dag_run_id): the request-scoped dag_bag
must reach dag.clear()."""
+ self.create_task_instances(session)
+ response = test_client.post(
+ "/dags/example_python_operator/clearTaskInstances",
+ json={"dry_run": True, "only_failed": False},
+ )
+
+ assert response.status_code == 200
+ assert mock_clear.call_count == 1
+ assert mock_clear.call_args.kwargs["dag_bag"] is
test_client.app.state.dag_bag
+
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.clear_task_instances")
+ @mock.patch("airflow.serialization.definitions.dag.SerializedDAG.clear")
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.task_instances.get_auth_manager")
+ def test_include_dependent_dags_filters_unauthorized_child_tis(
+ self, mock_get_auth_manager, mock_dag_clear, mock_clear_tis,
test_client, session
+ ):
+ """TIs from child DAGs the caller cannot edit must be excluded when
include_dependent_dags=True."""
+ import uuid
Review Comment:
Resolved in next commit.
--
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]
