This is an automated email from the ASF dual-hosted git repository.
kaxilnaik pushed a commit to branch v3-0-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit a6e6cb1736a08a75e3c4203390d541c2c8b32839
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Jun 6 13:43:21 2025 +0530
[v3-0-test] Remove logical_date check when validating inlets and outlets
(#51464) (#51469)
(cherry picked from commit 8865dbd4ca7d80aed7893688312e053c6d334022)
Co-authored-by: Wei Lee <[email protected]>
---
.../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 15cdb0a40ca..6c86ef38035 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
@@ -868,7 +868,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 49a8717c36a..13f0b8df846 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
@@ -2142,7 +2142,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(
@@ -2154,7 +2161,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")
@@ -2175,7 +2182,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(
@@ -2184,7 +2198,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")
