Block empty "chosen_options" from API

[Lee-W]

Body

as title (probably worth checking similar on the HITLOperator end as well)

Committer

• I acknowledge that I am a maintainer/committer of the Apache Airflow project.

[sjyangkevin]

I think it will be helpful for adding and testing this enhancement if we could get some context for how to properly test it 🤔. I assume that we can action on a running HITL task by using the endpoint below to provide an action. For example, I want to choose an option defined in the wait_for_option task (assume testing with the example DAG). How we can specify the parameters in the request body to properly trigger the task. If I sent out this request without making any change to the request body, I will get a 409 conflict error. The question is more like what are the steps we can follow to do a manual functional test using the DAG. Thanks!

[Lee-W]

here is an example code I use to test. to reproduce this bug, we'll need to send "chosen_options": [].

import requests

# manual input required
dag_run_id = <Dag run ID>
chosen_options = ["option 2"]
multiple_chosen_options = ["option 4", "option 6"]

params_input = {"information": "test info"}

# ---

host = "http://localhost:28080"
dag_id = "example_hitl_operator"


# authentication
token_endpoint = "/auth/token"
resp = requests.post(f"{host}{token_endpoint}", json={"username": "admin", "password": "admin"})
token = resp.json()["access_token"]
headers = {"Authorization": f"Bearer {token}"}
     
# get all hitl details                
get_all_hitl_details_endpoint = "/api/v2/hitlDetails/"
resp = requests.get(f"{host}{get_all_hitl_details_endpoint}", headers=headers)
resp.json()

# get hitl detail for task "wait_for_option"
task_id = "wait_for_option"
hitl_ti_endpoint = f"/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
resp = requests.get(f"{host}{hitl_ti_endpoint}", headers=headers)
resp.json()

# response to "wait_for_option"
data = {"chosen_options": chosen_options}
resp = requests.patch(f"{host}{hitl_ti_endpoint}", headers=headers, json=data)
resp.json()

# response to "wait_for_input"
task_id = "wait_for_input"
hitl_ti_endpoint = f"/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
data = {"params_input": params_input, "chosen_options": ["OK"]}
resp = requests.patch(f"{host}{hitl_ti_endpoint}", headers=headers, json=data)
resp.json()

# response to "wait_for_multiple_options"
task_id = "wait_for_multiple_options"
hitl_ti_endpoint = f"/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
data = {"chosen_options": multiple_chosen_options}
resp = requests.patch(f"{host}{hitl_ti_endpoint}", headers=headers, json=data)
resp.json()

# response to "valid_input_and_options", need to wait for the tasks above
task_id = "valid_input_and_options"
hitl_ti_endpoint = f"/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
data = {"chosen_options": ["Approve"]}
resp = requests.patch(f"{host}{hitl_ti_endpoint}", headers=headers, json=data)
resp.json()

# resposne to "choose_a_branch_to_run", need to wait for the tasks above
task_id = "choose_a_branch_to_run"
hitl_ti_endpoint = f"/api/v2/hitlDetails/{dag_id}/{dag_run_id}/{task_id}"
data = {"chosen_options": ["task_1"]}
resp = requests.patch(f"{host}{hitl_ti_endpoint}", headers=headers, json=data)
resp.json()

[sjyangkevin]

here is an example code I use to test. to reproduce this bug, we'll need to send "chosen_options": [].

Thanks a lot! It’s really helpful.