Lee-W commented on code in PR #53907: URL: https://github.com/apache/airflow/pull/53907#discussion_r2278052205
########## airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_hitl_shared_links.py: ########## @@ -0,0 +1,409 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +from __future__ import annotations + +import datetime +from typing import TYPE_CHECKING, Any +from unittest.mock import ANY + +import pytest +from fastapi import status +from sqlalchemy import select + +from airflow._shared.timezones.timezone import convert_to_utc +from airflow.api_fastapi.core_api.services.public.hitl_shared_links import ( + HITL_LINK_TYPE, + HITLSharedLinkConfig, + HITLSharedLinkData, + decode_shared_data, +) +from airflow.models.hitl import HITLDetail +from airflow.models.taskinstance import TaskInstance as TI + +from tests_common.pytest_plugin import time_machine +from tests_common.test_utils.config import conf_vars + +if TYPE_CHECKING: + from fastapi.testclient import TestClient + from sqlalchemy.orm import Session + + from tests_common.pytest_plugin import CreateTaskInstance, TaskInstance + +pytestmark = pytest.mark.db_test + +TEST_SERVER_PREFIX = "http://testserver/api/v2"; + +DAG_ID = "test_hitl_shared_link_dag" +DAG_RUN_ID = "run_id" +TASK_ID = "sample_hitl_task" + +CURRENT_TIME = convert_to_utc(datetime.datetime(2025, 8, 14, 0, 0, 0)) +ONE_DAY_BEFORE_CURRENT = CURRENT_TIME - datetime.timedelta(days=1) +ONE_DAY_AFTER_CURRENT = CURRENT_TIME + datetime.timedelta(days=1) + + [email protected] +def enabled(): + with conf_vars({("api", "hitl_enable_shared_links"): True}): + yield + + [email protected] +def sample_ti(create_task_instance: CreateTaskInstance, session: Session) -> TaskInstance: + ti = create_task_instance(dag_id=DAG_ID, run_id=DAG_RUN_ID, task_id=TASK_ID, session=session) + session.commit() + return ti + + [email protected]("map_index", [None, -1]) +class TestGenerateSharedLink: + """Tests for /hitlSharedLinks/generate endpoints""" + + def gen_endpoint(self, map_index: int | None) -> str: + base = f"/hitlSharedLinks/generate/{DAG_ID}/{DAG_RUN_ID}/{TASK_ID}" + return base if map_index is None else f"{base}/{map_index}" + + @pytest.mark.parametrize( + "link_type, chosen_options_in_payload, params_input_payload", + (("redirect", None, None), ("respond", ["option 1"], None)), + ) + @time_machine.travel(CURRENT_TIME, tick=False) + def test_should_respond_201( + self, + test_client: TestClient, + sample_ti: TaskInstance, + link_type: HITL_LINK_TYPE, + chosen_options_in_payload: list[str] | None, + params_input_payload: dict[str, Any] | None, + map_index: int | None, + ) -> None: + endpoint = self.gen_endpoint(map_index) + secret_key = test_client.app.state.secret_key # type: ignore[attr-defined] + + expected_data = HITLSharedLinkData( + ti_id=sample_ti.id, + dag_id=DAG_ID, + dag_run_id=DAG_RUN_ID, + task_id=TASK_ID, + map_index=map_index, + shared_link_config=HITLSharedLinkConfig( + link_type=link_type, + expires_at=ONE_DAY_AFTER_CURRENT.isoformat(), + chosen_options=chosen_options_in_payload or [], + params_input=params_input_payload or {}, + ), + ) + + resp = test_client.post( + endpoint, + json={ + "link_type": link_type, + "expires_at": None, + "chosen_options": chosen_options_in_payload, + "params_input": params_input_payload, + }, + ) + assert resp.status_code == status.HTTP_201_CREATED + url = resp.json()["url"] + token = url.split("/")[-1] + assert url.startswith(f"{TEST_SERVER_PREFIX}/hitlSharedLinks/{link_type}/") + assert decode_shared_data(secret_key=secret_key, token=token) == expected_data Review Comment: we should. Thanks I'll add it -- 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]
