This is an automated email from the ASF dual-hosted git repository.
weilee pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8865dbd4ca7 Remove logical_date check when validating inlets and
outlets (#51464)
8865dbd4ca7 is described below
commit 8865dbd4ca7d80aed7893688312e053c6d334022
Author: Wei Lee <weilee...@gmail.com>
AuthorDate: Fri Jun 6 15:08:37 2025 +0800
Remove logical_date check when validating inlets and outlets (#51464)
---
.../execution_api/routes/task_instances.py | 2 +-
.../versions/head/test_task_instances.py | 22 ++++++++++++++++++----
2 files changed, 19 insertions(+), 5 deletions(-)
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
index 0e7dadd3845..e22d0a5f34d 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/task_instances.py
@@ -899,7 +899,7 @@ def validate_inlets_and_outlets(
bind_contextvars(ti_id=ti_id_str)
ti = session.scalar(select(TI).where(TI.id == ti_id_str))
- if not ti or not ti.logical_date:
+ if not ti:
log.error("Task Instance not found")
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
index 67c1b2ca4af..3c107f61863 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_task_instances.py
@@ -2181,7 +2181,14 @@ class TestGetTaskStates:
class TestInvactiveInletsAndOutlets:
- def test_ti_inactive_inlets_and_outlets(self, client, dag_maker):
+ @pytest.mark.parametrize(
+ "logical_date",
+ [
+ datetime(2025, 6, 6, tzinfo=timezone.utc),
+ None,
+ ],
+ )
+ def test_ti_inactive_inlets_and_outlets(self, logical_date, client,
dag_maker):
"""Test the inactive assets in inlets and outlets can be found."""
with dag_maker("test_inlets_and_outlets"):
EmptyOperator(
@@ -2193,7 +2200,7 @@ class TestInvactiveInletsAndOutlets:
],
)
- dr = dag_maker.create_dagrun()
+ dr = dag_maker.create_dagrun(logical_date=logical_date)
task1_ti = dr.get_task_instance("task1")
response =
client.get(f"/execution/task-instances/{task1_ti.id}/validate-inlets-and-outlets")
@@ -2214,7 +2221,14 @@ class TestInvactiveInletsAndOutlets:
for asset in expected_inactive_assets:
assert asset in inactive_assets
- def test_ti_inactive_inlets_and_outlets_without_inactive_assets(self,
client, dag_maker):
+ @pytest.mark.parametrize(
+ "logical_date",
+ [
+ datetime(2025, 6, 6, tzinfo=timezone.utc),
+ None,
+ ],
+ )
+ def test_ti_inactive_inlets_and_outlets_without_inactive_assets(self,
logical_date, client, dag_maker):
"""Test the task without inactive assets in its inlets or outlets
returns empty list."""
with dag_maker("test_inlets_and_outlets_inactive"):
EmptyOperator(
@@ -2223,7 +2237,7 @@ class TestInvactiveInletsAndOutlets:
outlets=[Asset(name="outlet-name", uri="uri")],
)
- dr = dag_maker.create_dagrun()
+ dr = dag_maker.create_dagrun(logical_date=logical_date)
task1_ti = dr.get_task_instance("inactive_task1")
response =
client.get(f"/execution/task-instances/{task1_ti.id}/validate-inlets-and-outlets")
