nothingmin commented on code in PR #54416:
URL: https://github.com/apache/airflow/pull/54416#discussion_r2288659884
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -2312,10 +2311,103 @@ def
test_dag_run_with_future_or_past_flag_returns_400(self, test_client, session
response = test_client.post(f"/dags/{dag_id}/clearTaskInstances",
json=payload)
assert response.status_code == 400
assert (
- "Cannot use include_past or include_future when dag_run_id is
provided"
+ "Cannot use include_past or include_future with a manually
triggered DAG run (logical_date is None)."
in response.json()["detail"]
)
+ @pytest.mark.parametrize(
+ "flag, expected",
+ [
+ ("include_past", 2), # D0 ~ D1
+ ("include_future", 2), # D1 ~
+ ],
+ )
+ def test_with_dag_run_id_and_past_future_converts_to_date_range(self,
test_client, session, flag,
+ expected):
+ dag_id = "example_python_operator"
+ task_instances = [
+ {"logical_date": DEFAULT_DATETIME_1, "state": State.FAILED}, # D0
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=1),
"state": State.FAILED}, # D1
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=2),
"state": State.FAILED}, # D2
+ ]
+ self.create_task_instances(
+ session, dag_id=dag_id, task_instances=task_instances,
update_extras=False
+ )
+ payload = {
+ "dry_run": True,
+ "only_failed": True,
+ "dag_run_id": "TEST_DAG_RUN_ID_1",
+ flag: True,
+ }
+ resp = test_client.post(f"/dags/{dag_id}/clearTaskInstances",
json=payload)
+ assert resp.status_code == 200
+ assert resp.json()["total_entries"] == expected # include_past =>
D0,D1 / include_future => D1,D2
+
+ def test_with_dag_run_id_and_both_past_and_future_means_full_range(self,
test_client, session):
+ dag_id = "example_python_operator"
+ task_instances = [
+ {"logical_date": DEFAULT_DATETIME_1 - dt.timedelta(days=1),
"state": State.FAILED}, # D0
+ {"logical_date": DEFAULT_DATETIME_1, "state": State.FAILED}, #D1
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=1),
"state": State.FAILED}, # D2
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=2),
"state": State.FAILED}, # D3
+ {"logical_date": None, "state": State.FAILED}, # D4
+ ]
+ self.create_task_instances(
+ session, dag_id=dag_id, task_instances=task_instances,
update_extras=False
+ )
+ payload = {
+ "dry_run": True,
+ "only_failed": False,
+ "dag_run_id": "TEST_DAG_RUN_ID_1", # D1
+ "include_past": True,
+ "include_future": True,
+ }
+ resp = test_client.post(f"/dags/{dag_id}/clearTaskInstances",
json=payload)
+ assert resp.status_code == 200
+ assert resp.json()["total_entries"] == 5 #D0 ~ #D4
+
+ def test_with_dag_run_id_only_uses_run_id_based_clearing(self,
test_client, session):
+ dag_id = "example_python_operator"
+ task_instances = [
+ {"logical_date": DEFAULT_DATETIME_1, "state": State.SUCCESS}, # D0
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=1),
"state": State.FAILED}, # D1
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=2),
"state": State.SUCCESS}, # D2
+ ]
+ self.create_task_instances(
+ session, dag_id=dag_id, task_instances=task_instances,
update_extras=False
+ )
+ payload = {
+ "dry_run": True,
+ "only_failed": True,
+ "dag_run_id": "TEST_DAG_RUN_ID_1",
+ }
+ resp = test_client.post(f"/dags/{dag_id}/clearTaskInstances",
json=payload)
+ assert resp.status_code == 200
+ a = resp.json()
+ assert resp.json()["total_entries"] == 1
+ assert resp.json()["task_instances"][0]["logical_date"] ==
(DEFAULT_DATETIME_1 + dt.timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%SZ") #D1
+
+ @pytest.mark.parametrize("flag", ["include_future", "include_past"])
+ def test_manual_run_with_none_logical_date_returns_400_kept(self,
test_client, session, flag):
Review Comment:
It was just a typo, so I deleted it. Thanks!
--
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]
