[airflow] branch main updated: Add test case for clearTaskInstance call with invalid Task IDs. (#22894)

[ur...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

uranusjr pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git


The following commit(s) were added to refs/heads/main by this push:
     new 1a8b8f521c Add test case for clearTaskInstance call with invalid Task IDs. (#22894)
1a8b8f521c is described below

commit 1a8b8f521c887716d7e0c987a58e8e5c3b62bdaa
Author: Karthikeyan Singaravelan <ti...@gmail.com>
AuthorDate: Mon Apr 11 14:35:12 2022 +0530

    Add test case for clearTaskInstance call with invalid Task IDs. (#22894)
---
 .../endpoints/test_task_instance_endpoint.py         | 20 ++++++++++++++++++++
 1 file changed, 20 insertions(+)

diff --git a/tests/api_connexion/endpoints/test_task_instance_endpoint.py b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
index 6e34d24cd6..a7ffd6e803 100644
--- a/tests/api_connexion/endpoints/test_task_instance_endpoint.py
+++ b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
@@ -1026,6 +1026,26 @@ class TestPostClearTaskInstances(TestTaskInstanceEndpoint):
             [], session, dag=self.app.dag_bag.get_dag(dag_id), dag_run_state=State.QUEUED
         )
 
+    def test_clear_taskinstance_is_called_with_invalid_task_ids(self, session):
+        """Test that dagrun is running when invalid task_ids are passed to clearTaskInstances API."""
+        dag_id = "example_python_operator"
+        tis = self.create_task_instances(session)
+        dagrun = tis[0].get_dagrun()
+        assert dagrun.state == 'running'
+
+        payload = {"dry_run": False, "reset_dag_runs": True, "task_ids": [""]}
+        self.app.dag_bag.sync_to_db()
+        response = self.client.post(
+            f"/api/v1/dags/{dag_id}/clearTaskInstances",
+            environ_overrides={"REMOTE_USER": "test"},
+            json=payload,
+        )
+        assert response.status_code == 200
+
+        dagrun.refresh_from_db()
+        assert dagrun.state == 'running'
+        assert all(ti.state == 'running' for ti in tis)
+
     def test_should_respond_200_with_reset_dag_run(self, session):
         dag_id = "example_python_operator"
         payload = {