pierrejeambrun commented on code in PR #54416:
URL: https://github.com/apache/airflow/pull/54416#discussion_r2315619302
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -2395,10 +2394,81 @@ 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), # T0 ~ T1
+ ("include_future", 2), # T1 ~ T2
+ ],
+ )
+ 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}, # T0
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=1),
"state": State.FAILED}, # T1
+ {"logical_date": DEFAULT_DATETIME_1 + dt.timedelta(days=2),
"state": State.FAILED}, # T2
+ ]
+ 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 =>
T0,1 / include_future => T1,T2
Review Comment:
```suggestion
assert resp.json()["total_entries"] == expected # include_past =>
T0,T1 / include_future => T1,T2
```
##########
airflow-core/src/airflow/api_fastapi/core_api/routes/public/task_instances.py:
##########
@@ -707,13 +707,13 @@ def post_clear_task_instances(
raise HTTPException(status.HTTP_404_NOT_FOUND, error_message)
# Get the specific dag version:
dag = get_dag_for_run(dag_bag, dag_run, session)
- if past or future:
+ if (past or future) and dag_run.logical_date is None:
raise HTTPException(
status.HTTP_400_BAD_REQUEST,
- "Cannot use include_past or include_future when dag_run_id is
provided because logical_date is not applicable.",
+ "Cannot use include_past or include_future with a manually
triggered DAG run (logical_date is None).",
)
- body.start_date = dag_run.logical_date if dag_run.logical_date is not
None else None
- body.end_date = dag_run.logical_date if dag_run.logical_date is not
None else None
+ body.start_date = dag_run.logical_date
+ body.end_date = dag_run.logical_date
Review Comment:
We probably still want:
```
body.start_date = dag_run.logical_date if dag_run.logical_date is not None
else None
body.end_date = dag_run.logical_date if dag_run.logical_date is not None
else None
```
Because we could reach that with no past / future but a logical date that is
`false`. (past=None, future=None, and logical date is None).
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3630,50 +3700,49 @@ def
test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte
"new_state,expected_status_code,expected_json,set_ti_state_call_count",
[
(
- "failed",
- 200,
- {
- "task_instances": [
- {
- "dag_id": "example_python_operator",
- "dag_display_name": "example_python_operator",
- "dag_version": mock.ANY,
- "dag_run_id": "TEST_DAG_RUN_ID",
- "logical_date": "2020-01-01T00:00:00Z",
- "task_id": "print_the_context",
- "duration": 10000.0,
- "end_date": "2020-01-03T00:00:00Z",
- "executor": None,
- "executor_config": "{}",
- "hostname": "",
- "id": mock.ANY,
- "map_index": -1,
- "max_tries": 0,
- "note": "placeholder-note",
- "operator": "PythonOperator",
- "operator_name": "PythonOperator",
- "pid": 100,
- "pool": "default_pool",
- "pool_slots": 1,
- "priority_weight": 9,
- "queue": "default_queue",
- "queued_when": None,
- "scheduled_when": None,
- "start_date": "2020-01-02T00:00:00Z",
- "state": "running",
- "task_display_name": "print_the_context",
- "try_number": 0,
- "unixname": getuser(),
- "rendered_fields": {},
- "rendered_map_index": None,
- "run_after": "2020-01-01T00:00:00Z",
- "trigger": None,
- "triggerer_job": None,
- }
- ],
- "total_entries": 1,
- },
- 1,
+ "failed",
+ 200,
+ {
+ "task_instances": [
+ {
+ "dag_id": "example_python_operator",
+ "dag_display_name": "example_python_operator",
+ "dag_version": mock.ANY,
+ "dag_run_id": "TEST_DAG_RUN_ID",
+ "logical_date": "2020-01-01T00:00:00Z",
+ "task_id": "print_the_context",
+ "duration": 10000.0,
+ "end_date": "2020-01-03T00:00:00Z",
+ "executor": None,
+ "executor_config": "{}",
+ "hostname": "",
+ "id": mock.ANY,
+ "map_index": -1,
+ "max_tries": 0,
+ "note": "placeholder-note",
+ "operator": "PythonOperator",
Review Comment:
here you're missing operator_name (bad rebase I believe)
##########
airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_task_instances.py:
##########
@@ -3630,50 +3700,49 @@ def
test_should_raise_422_for_invalid_task_instance_state(self, payload, expecte
"new_state,expected_status_code,expected_json,set_ti_state_call_count",
[
(
- "failed",
- 200,
- {
- "task_instances": [
- {
- "dag_id": "example_python_operator",
- "dag_display_name": "example_python_operator",
- "dag_version": mock.ANY,
- "dag_run_id": "TEST_DAG_RUN_ID",
- "logical_date": "2020-01-01T00:00:00Z",
- "task_id": "print_the_context",
- "duration": 10000.0,
- "end_date": "2020-01-03T00:00:00Z",
- "executor": None,
- "executor_config": "{}",
- "hostname": "",
- "id": mock.ANY,
- "map_index": -1,
- "max_tries": 0,
- "note": "placeholder-note",
- "operator": "PythonOperator",
- "operator_name": "PythonOperator",
- "pid": 100,
- "pool": "default_pool",
- "pool_slots": 1,
- "priority_weight": 9,
- "queue": "default_queue",
- "queued_when": None,
- "scheduled_when": None,
- "start_date": "2020-01-02T00:00:00Z",
- "state": "running",
- "task_display_name": "print_the_context",
- "try_number": 0,
- "unixname": getuser(),
- "rendered_fields": {},
- "rendered_map_index": None,
- "run_after": "2020-01-01T00:00:00Z",
- "trigger": None,
- "triggerer_job": None,
- }
- ],
- "total_entries": 1,
- },
- 1,
+ "failed",
+ 200,
+ {
+ "task_instances": [
+ {
+ "dag_id": "example_python_operator",
+ "dag_display_name": "example_python_operator",
+ "dag_version": mock.ANY,
+ "dag_run_id": "TEST_DAG_RUN_ID",
+ "logical_date": "2020-01-01T00:00:00Z",
+ "task_id": "print_the_context",
+ "duration": 10000.0,
+ "end_date": "2020-01-03T00:00:00Z",
+ "executor": None,
+ "executor_config": "{}",
+ "hostname": "",
+ "id": mock.ANY,
+ "map_index": -1,
+ "max_tries": 0,
+ "note": "placeholder-note",
+ "operator": "PythonOperator",
+ "pid": 100,
+ "pool": "default_pool",
+ "pool_slots": 1,
+ "priority_weight": 9,
+ "queue": "default_queue",
+ "queued_when": None,
+ "scheduled_when": None,
+ "start_date": "2020-01-02T00:00:00Z",
+ "state": "running",
+ "task_display_name": "print_the_context",
+ "try_number": 0,
+ "unixname": getuser(),
Review Comment:
Seems unrelated
--
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]
