Lee-W commented on code in PR #54827:
URL: https://github.com/apache/airflow/pull/54827#discussion_r2303242379
##########
providers/standard/tests/unit/standard/operators/test_hitl.py:
##########
@@ -238,6 +264,170 @@ def test_validate_params_input_with_invalid_input(self)
-> None:
},
)
+ @pytest.mark.parametrize(
+ "options, params_input, expected_query_string",
+ [
+ (None, None, "?map_index=-1"),
+ ("1", None, "?_options=1&map_index=-1"),
+ (["1", "2"], None, "?_options=1%2C2&map_index=-1"),
+ (None, {"input_1": 123}, "?input_1=123&map_index=-1"),
+ (
+ ["3", "4", "5"],
+ {"input_1": 123123, "input_2": 345345},
+
"?_options=3%2C4%2C5&input_1=123123&input_2=345345&map_index=-1",
+ ),
+ ],
+ ids=[
+ "empty",
+ "single-option",
+ "multiple-options",
+ "single-param-input",
+ "multiple-options-and-param-inputs",
+ ],
+ )
+ @pytest.mark.parametrize("base_url", ["http://test";, "http://test_2:8080";])
+ def test_generate_link_to_ui(
+ self,
+ dag_maker: DagMaker,
+ base_url: str,
+ options: list[str] | None,
+ params_input: dict[str, Any] | None,
+ expected_query_string: str,
+ ) -> None:
+ with dag_maker("test_dag"):
+ task = HITLOperator(
+ task_id="hitl_test",
+ subject="This is subject",
+ options=["1", "2", "3", "4", "5"],
+ body="This is body",
+ defaults=["1"],
+ respondents="test",
+ multiple=True,
+ params=ParamsDict({"input_1": 1, "input_2": 2, "input_3": 3}),
+ )
+ dr = dag_maker.create_dagrun()
+ ti = dag_maker.run_ti(task.task_id, dr)
+
+ expected_url = (
+
f"{base_url}/dags/test_dag/runs/test/tasks/hitl_test/required_actions{expected_query_string}"
+ )
+
+ url = task.generate_link_to_ui(
+ task_instance=ti,
+ base_url=base_url,
+ options=options,
+ params_input=params_input,
+ )
+ assert url == expected_url
+
+ @pytest.mark.parametrize(
+ "options, params_input, expected_query_string",
+ [
+ (None, None, "?map_index=-1"),
+ ("1", None, "?_options=1&map_index=-1"),
+ (["1", "2"], None, "?_options=1%2C2&map_index=-1"),
+ (None, {"input_1": 123}, "?input_1=123&map_index=-1"),
+ (
+ ["3", "4", "5"],
+ {"input_1": 123123, "input_2": 345345},
+
"?_options=3%2C4%2C5&input_1=123123&input_2=345345&map_index=-1",
+ ),
+ ],
+ ids=[
+ "empty",
+ "single-option",
+ "multiple-options",
+ "single-param-input",
+ "multiple-options-and-param-inputs",
+ ],
+ )
+ @conf_vars({("api", "base_url"): "http://localhost:8080/"})
+ def test_generate_link_fall_back_to_conf_api_base_url(
+ self,
+ dag_maker: DagMaker,
+ options: list[str] | None,
+ params_input: dict[str, Any] | None,
+ expected_query_string: str,
+ ) -> None:
+ with dag_maker("test_dag"):
+ task = HITLOperator(
+ task_id="hitl_test",
+ subject="This is subject",
+ options=["1", "2", "3", "4", "5"],
+ body="This is body",
+ defaults=["1"],
+ respondents="test",
+ multiple=True,
+ params=ParamsDict({"input_1": 1, "input_2": 2, "input_3": 3}),
+ )
+ dr = dag_maker.create_dagrun()
+ ti = dag_maker.run_ti(task.task_id, dr)
+
+ expected_url =
f"http://localhost:8080/dags/test_dag/runs/test/tasks/hitl_test/required_actions{expected_query_string}";
+
+ url = task.generate_link_to_ui(
+ task_instance=ti,
+ options=options,
+ params_input=params_input,
+ )
+ assert url == expected_url
+
+ @pytest.mark.parametrize(
+ "options, params_input, expected_err_msg",
+ [
+ ([100, "2", 30000], None, "options {.*} are not valid options"),
+ (
+ None,
+ {"input_not_exist": 123, "no_such_key": 123},
+ "params {.*} are not valid params",
+ ),
+ ],
+ )
+ def test_generate_link_to_ui_with_invalid_input(
+ self,
+ dag_maker: DagMaker,
+ options: list[str] | None,
+ params_input: dict[str, Any] | None,
+ expected_err_msg: str,
+ ) -> None:
+ with dag_maker("test_dag"):
+ task = HITLOperator(
+ task_id="hitl_test",
+ subject="This is subject",
+ options=["1", "2", "3", "4", "5"],
+ body="This is body",
+ defaults=["1"],
+ respondents="test",
+ multiple=True,
+ params=ParamsDict({"input_1": 1, "input_2": 2, "input_3": 3}),
+ )
+ dr = dag_maker.create_dagrun()
+ ti = dag_maker.run_ti(task.task_id, dr)
+
+ with pytest.raises(ValueError, match=expected_err_msg):
+ task.generate_link_to_ui(task_instance=ti, options=options,
params_input=params_input)
+
+ def test_generate_link_to_ui_without_base_url(
+ self,
+ dag_maker: DagMaker,
+ ) -> None:
+ with dag_maker("test_dag"):
+ task = HITLOperator(
+ task_id="hitl_test",
+ subject="This is subject",
+ options=["1", "2", "3", "4", "5"],
+ body="This is body",
+ defaults=["1"],
+ respondents="test",
+ multiple=True,
+ params=ParamsDict({"input_1": 1, "input_2": 2, "input_3": 3}),
+ )
+ dr = dag_maker.create_dagrun()
+ ti = dag_maker.run_ti(task.task_id, dr)
Review Comment:
I later make it `hitl_task_and_ti_for_generating_link`. Not a fan of making
the return type complicate, but I guess that's the best we can do now 🤔
--
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]
