kaxil commented on code in PR #67273:
URL: https://github.com/apache/airflow/pull/67273#discussion_r3439589449
##########
airflow-core/src/airflow/api_fastapi/execution_api/routes/dag_runs.py:
##########
@@ -30,6 +30,7 @@
from airflow.api_fastapi.common.db.common import SessionDep
from airflow.api_fastapi.common.types import UtcDateTime
from airflow.api_fastapi.compat import HTTP_422_UNPROCESSABLE_CONTENT
+from airflow.api_fastapi.core_api.services.public.common import
resolve_run_on_latest_version
Review Comment:
Can we avoid importing from the public REST service layer here? The
execution API is worker-facing, and its existing `core_api` dependencies are
only `core_api.base` and `core_api.openapi.exceptions` (shared leaf modules).
`core_api.services.public.common` is the REST service tier, and it imports the
`Bulk*` REST datamodels at module level, so this one-line import now pulls the
public bulk-operations layer into the worker API.
`resolve_run_on_latest_version` has no REST dependency of its own (it only
touches `conf` and `SerializedDagModel`), so it doesn't really belong in the
public-service module. Moving it to `airflow.api_fastapi.common` (which both
`core_api` and `execution_api` already depend on) and importing it from there
in both this endpoint and the public clear endpoint keeps it shared without
crossing the execution -> core_api.services boundary.
##########
airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py:
##########
@@ -322,6 +324,27 @@ def test_dag_run_not_found(self, client):
assert response.status_code == 404
+ def test_dag_run_clear_invokes_resolver(self, client, session, dag_maker):
+ """Clearing invokes resolve_run_on_latest_version with no explicit
override."""
+ dag_id = "test_clear_invokes_resolver"
+ run_id = "test_run_id"
+
+ with dag_maker(dag_id=dag_id, session=session, serialized=True):
+ EmptyOperator(task_id="test_task")
+ dag_maker.create_dagrun(run_id=run_id, state=DagRunState.SUCCESS)
+ session.commit()
+
+ with mock.patch(
+
"airflow.api_fastapi.execution_api.routes.dag_runs.resolve_run_on_latest_version",
+ return_value=False,
+ ) as mock_resolver:
+ response =
client.post(f"/execution/dag-runs/{dag_id}/{run_id}/clear")
+
+ assert response.status_code == 204
+ mock_resolver.assert_called_once()
Review Comment:
This asserts the resolver is *called* with `None`, but not that its return
value reaches `dag.clear()`. The resolver is mocked to return `False`, yet
nothing checks that `False` is what `dag.clear` receives, so an implementation
that computes `resolved_run_on_latest` and then calls
`dag.clear(run_id=run_id)` (dropping it) would still pass. That dropped-value
case is exactly the regression this PR guards against. Consider spying on
`SerializedDAG.clear` (autospec) and asserting it gets
`run_on_latest_version=<the resolved value>`, so the test pins the data flow
and not just the call.
--
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]
