Allow skipped task state task_instance_schema.py (#31421)

---------

Co-authored-by: Hussein Awala <hussein@awala.fr>

Author: JoranDox

airflow/api_connexion/openapi/v1.yaml

@@ -4265,6 +4265,7 @@ components:
           enum:
             - success
             - failed
+            - skipped
 
     UpdateTaskInstance:
       type: object
@@ -4282,6 +4283,8 @@ components:
           enum:
             - success
             - failed
+            - skipped
+
     SetTaskInstanceNote:
       type: object
       required:

airflow/api_connexion/schemas/task_instance_schema.py

@@ -30,7 +30,7 @@
 from airflow.api_connexion.schemas.trigger_schema import TriggerSchema
 from airflow.models import SlaMiss, TaskInstance
 from airflow.utils.helpers import exactly_one
-from airflow.utils.state import State
+from airflow.utils.state import TaskInstanceState
 
 
 class TaskInstanceSchema(SQLAlchemySchema):
@@ -158,7 +158,12 @@ class SetTaskInstanceStateFormSchema(Schema):
     include_downstream = fields.Boolean(required=True)
     include_future = fields.Boolean(required=True)
     include_past = fields.Boolean(required=True)
-    new_state = TaskInstanceStateField(required=True, validate=validate.OneOf([State.SUCCESS, State.FAILED]))
+    new_state = TaskInstanceStateField(
+        required=True,
+        validate=validate.OneOf(
+            [TaskInstanceState.SUCCESS, TaskInstanceState.FAILED, TaskInstanceState.SKIPPED]
+        ),
+    )
 
     @validates_schema
     def validate_form(self, data, **kwargs):
@@ -171,7 +176,12 @@ class SetSingleTaskInstanceStateFormSchema(Schema):
     """Schema for handling the request of updating state of a single task instance."""
 
     dry_run = fields.Boolean(dump_default=True)
-    new_state = TaskInstanceStateField(required=True, validate=validate.OneOf([State.SUCCESS, State.FAILED]))
+    new_state = TaskInstanceStateField(
+        required=True,
+        validate=validate.OneOf(
+            [TaskInstanceState.SUCCESS, TaskInstanceState.FAILED, TaskInstanceState.SKIPPED]
+        ),
+    )
 
 
 class TaskInstanceReferenceSchema(Schema):

airflow/www/static/js/types/api-generated.ts

@@ -1874,7 +1874,7 @@ export interface components {
        * @description Expected new state.
        * @enum {string}
        */
-      new_state?: "success" | "failed";
+      new_state?: "success" | "failed" | "skipped";
     };
     UpdateTaskInstance: {
       /**
@@ -1888,7 +1888,7 @@ export interface components {
        * @description Expected new state.
        * @enum {string}
        */
-      new_state?: "success" | "failed";
+      new_state?: "success" | "failed" | "skipped";
     };
     SetTaskInstanceNote: {
       /** @description The custom note to set for this Task Instance. */

tests/api_connexion/endpoints/test_task_instance_endpoint.py

@@ -1803,14 +1803,16 @@ def test_should_raise_404_not_found_task(self):
                     "dry_run": True,
                     "new_state": "failede",
                 },
-                f"'failede' is not one of ['{State.SUCCESS}', '{State.FAILED}'] - 'new_state'",
+                f"'failede' is not one of ['{State.SUCCESS}', '{State.FAILED}', '{State.SKIPPED}']"
+                " - 'new_state'",
             ),
             (
                 {
                     "dry_run": True,
                     "new_state": "queued",
                 },
-                f"'queued' is not one of ['{State.SUCCESS}', '{State.FAILED}'] - 'new_state'",
+                f"'queued' is not one of ['{State.SUCCESS}', '{State.FAILED}', '{State.SKIPPED}']"
+                " - 'new_state'",
             ),
         ],
     )