jason810496 commented on code in PR #54827:
URL: https://github.com/apache/airflow/pull/54827#discussion_r2300782156
##########
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:
How about having another fixture to setup the Dag for generating link for UI
and yield the `HITLOperator instance`, `DagRun` and `TaskInstance`?
It seems the Dags in these test cases are same.
--
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: commits-unsubscr...@airflow.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
