This is an automated email from the ASF dual-hosted git repository.
potiuk 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 325360c2b92 Reject reserved serialization keys in Dag run conf (#74834)
325360c2b92 is described below
commit 325360c2b9252d9b0fe707f2fdcc6b6882504ca8
Author: Jarek Potiuk <[email protected]>
AuthorDate: Mon Oct 19 16:31:49 2026 +0200
Reject reserved serialization keys in Dag run conf (#74834)
* Reject reserved serialization keys in Dag run conf
XCom values and Human-in-the-loop params already reject keys reserved by
the serialization format. Apply the same check to Dag run conf where it
enters through the REST API (trigger, asset materialize, backfill) and
the Execution API trigger route. A triggerer Deadline callback whose
context still holds reserved keys in either encoding is marked failed
instead of queued, so the scheduler loop keeps handling other deadlines.
Generated-by: Claude Opus 5 following the guidelines at
https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions
* Move the conf reserved-key check next to find_reserved_keys
The backfill and Execution API datamodels import it too, so keeping it in
dag_run.py made two unrelated models depend on the Dag run module for a shared
validation helper.
Generated-by: Claude Opus 5
---
airflow-core/newsfragments/74834.bugfix.rst | 1 +
.../api_fastapi/core_api/datamodels/backfills.py | 8 ++-
.../api_fastapi/core_api/datamodels/common.py | 14 +++++
.../api_fastapi/core_api/datamodels/dag_run.py | 8 ++-
.../api_fastapi/execution_api/datamodels/dagrun.py | 8 ++-
airflow-core/src/airflow/models/deadline.py | 33 ++++++++++-
.../core_api/datamodels/test_dag_run.py | 66 +++++++++++++++++++++-
.../core_api/routes/public/test_assets.py | 15 +++++
.../core_api/routes/public/test_backfills.py | 27 +++++++++
.../core_api/routes/public/test_dag_run.py | 25 ++++++++
.../execution_api/versions/head/test_dag_runs.py | 21 +++++++
airflow-core/tests/unit/jobs/test_scheduler_job.py | 59 +++++++++++++++++++
airflow-core/tests/unit/models/test_deadline.py | 57 ++++++++++++++++++-
13 files changed, 334 insertions(+), 8 deletions(-)
diff --git a/airflow-core/newsfragments/74834.bugfix.rst
b/airflow-core/newsfragments/74834.bugfix.rst
new file mode 100644
index 00000000000..0382f2b6364
--- /dev/null
+++ b/airflow-core/newsfragments/74834.bugfix.rst
@@ -0,0 +1 @@
+Reject Dag run ``conf`` holding reserved serialization keys such as
``__classname__`` or ``__type``, as XCom and Human-in-the-loop values already
do. The trigger Dag run, materialize asset and create backfill endpoints, and
the Execution API trigger Dag run endpoint, now return 422 for such a payload;
string values in ``conf`` are not inspected. A Deadline Alert ``AsyncCallback``
whose context still holds these keys is marked failed instead of queued, and
the scheduler moves on to other [...]
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
index 9788546a401..3833825479d 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/backfills.py
@@ -20,9 +20,10 @@ from __future__ import annotations
from collections.abc import Iterable
from datetime import datetime
-from pydantic import AliasPath, Field, NonNegativeInt
+from pydantic import AliasPath, Field, NonNegativeInt, field_validator
from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
+from airflow.api_fastapi.core_api.datamodels.common import
check_conf_reserved_keys
from airflow.models.backfill import ReprocessBehavior
from airflow.utils.state import DagRunState
@@ -53,6 +54,11 @@ class BackfillPostBody(StrictBaseModel):
"Ignored by the dry-run endpoint.",
)
+ @field_validator("dag_run_conf")
+ @classmethod
+ def _check_reserved_keys(cls, dag_run_conf: dict | None) -> dict | None:
+ return check_conf_reserved_keys(dag_run_conf, root="dag_run_conf")
+
class BackfillResponse(BaseModel):
"""Base serializer for Backfill."""
diff --git a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py
index 7dc55862e67..5bae6c5888c 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/common.py
@@ -83,6 +83,20 @@ def find_reserved_keys(
return walk(value, root)
+def check_conf_reserved_keys(conf: dict | None, *, root: str) -> dict | None:
+ """Reject a Dag run conf holding reserved serialization keys, as XCom and
HITL values do."""
+ # Conf reaches code that runs a full serde decode (e.g. the Deadline
callback context). Unlike
+ # XCom, nothing parses a string conf value back into a container, so
strings stay unchecked.
+ found = find_reserved_keys(conf, root=root)
+ if found is not None:
+ path, keys = found
+ raise ValueError(
+ f"{path} contains reserved serialization keys: {', '.join(keys)}. "
+ "These keys are reserved for internal use."
+ )
+ return conf
+
+
# Asset Scheduling Expression Data Models
#
# These mirror the JSON produced by ``BaseAsset.as_expression()`` (see
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
index a959d7fca19..f388bef7940 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/datamodels/dag_run.py
@@ -22,10 +22,11 @@ from datetime import datetime
from enum import Enum
from typing import TYPE_CHECKING, Any
-from pydantic import AliasPath, AwareDatetime, Field, NonNegativeInt,
model_validator
+from pydantic import AliasPath, AwareDatetime, Field, NonNegativeInt,
field_validator, model_validator
from airflow._shared.timezones import timezone
from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
+from airflow.api_fastapi.core_api.datamodels.common import
check_conf_reserved_keys
from airflow.api_fastapi.core_api.datamodels.dag_versions import
DagVersionResponse
from airflow.timetables.base import DataInterval
from airflow.utils.state import DagRunState
@@ -244,6 +245,11 @@ class TriggerDAGRunPostBody(StrictBaseModel):
"to complete. Requires the same permission as pausing the Dag.",
)
+ @field_validator("conf")
+ @classmethod
+ def _check_reserved_keys(cls, conf: dict | None) -> dict | None:
+ return check_conf_reserved_keys(conf, root="conf")
+
@model_validator(mode="after")
def check_data_intervals(self):
if (self.data_interval_start is None) != (self.data_interval_end is
None):
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/dagrun.py
b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/dagrun.py
index dd1f1aeb866..15631a2cb35 100644
--- a/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/dagrun.py
+++ b/airflow-core/src/airflow/api_fastapi/execution_api/datamodels/dagrun.py
@@ -17,10 +17,11 @@
from __future__ import annotations
-from pydantic import Field
+from pydantic import Field, field_validator
from airflow.api_fastapi.common.types import UtcDateTime
from airflow.api_fastapi.core_api.base import BaseModel, StrictBaseModel
+from airflow.api_fastapi.core_api.datamodels.common import
check_conf_reserved_keys
from airflow.utils.state import DagRunState
@@ -34,6 +35,11 @@ class TriggerDAGRunPayload(StrictBaseModel):
partition_key: str | None = None
note: str | None = None
+ @field_validator("conf")
+ @classmethod
+ def _check_reserved_keys(cls, conf: dict) -> dict | None:
+ return check_conf_reserved_keys(conf, root="conf")
+
class DagRunStateResponse(BaseModel):
"""Schema for DAG Run State response."""
diff --git a/airflow-core/src/airflow/models/deadline.py
b/airflow-core/src/airflow/models/deadline.py
index b9c212dc834..b2c80d1b576 100644
--- a/airflow-core/src/airflow/models/deadline.py
+++ b/airflow-core/src/airflow/models/deadline.py
@@ -289,10 +289,33 @@ class Deadline(Base):
data["kwargs"] = kwargs
return data
+ callback_failed = False
if isinstance(self.callback, TriggererCallback):
- self.callback.data = callback_data_with_context()
+ from airflow.api_fastapi.core_api.datamodels.common import
find_reserved_keys
- self.callback.queue(session=session)
+ data = callback_data_with_context()
+ # The triggerer decodes the context with full serde, which
rebuilds a value in either
+ # encoding as an object, and strict serialize raises on the
current one. Fail the
+ # callback instead of queueing it: raising here would abort the
scheduler loop, and the
+ # deadline would be picked up and raise again on every restart.
+ if found := find_reserved_keys(data["kwargs"]["context"],
root="context"):
+ _, keys = found
+ logger.error(
+ "Not queueing the callback of deadline %s for Dag %s, run
%s: its context holds "
+ "reserved serialization keys %s",
+ self.id,
+ self.dagrun.dag_id,
+ self.dagrun.run_id,
+ ", ".join(keys),
+ )
+ self.callback.state = CallbackState.FAILED
+ self.callback.output = (
+ f"Not queued: the callback context holds reserved
serialization keys {', '.join(keys)}"
+ )
+ callback_failed = True
+ else:
+ self.callback.data = data
+ self.callback.queue(session=session)
session.add(self.callback)
session.flush()
@@ -324,6 +347,12 @@ class Deadline(Base):
{"dag_id": self.dagrun.dag_id, "dagrun_id":
self.dagrun.run_id, "team_name": team_name}
),
)
+ if callback_failed:
+ stats.incr(
+ **self.callback.get_metric_info(
+ CallbackState.FAILED, self.callback.output,
team_name=team_name
+ )
+ )
class ReferenceModels:
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_dag_run.py
index 4fd4adb845b..d1f841041ee 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/datamodels/test_dag_run.py
@@ -16,11 +16,20 @@
# under the License.
from __future__ import annotations
+import json
from datetime import UTC, datetime
import pytest
+from pydantic import ValidationError
-from airflow.api_fastapi.core_api.datamodels.dag_run import
BulkDAGRunClearBody, ClearPartitionsBody
+from airflow._shared.serialization import FORBIDDEN_XCOM_KEYS
+from airflow.api_fastapi.core_api.datamodels.assets import MaterializeAssetBody
+from airflow.api_fastapi.core_api.datamodels.backfills import BackfillPostBody
+from airflow.api_fastapi.core_api.datamodels.dag_run import (
+ BulkDAGRunClearBody,
+ ClearPartitionsBody,
+ TriggerDAGRunPostBody,
+)
_DATE_A = datetime(2024, 1, 1, tzinfo=UTC)
_DATE_B = datetime(2024, 1, 31, tzinfo=UTC)
@@ -67,3 +76,58 @@ class TestDateWindowOrder:
def test_accepts_start_before_end(self, model_cls):
body = model_cls(partition_date_start=_DATE_A,
partition_date_end=_DATE_B)
assert body.partition_date_start < body.partition_date_end
+
+
+_CONF_BODIES = [
+ pytest.param(TriggerDAGRunPostBody, "conf", {"logical_date": None},
id="trigger"),
+ pytest.param(MaterializeAssetBody, "conf", {}, id="materialize"),
+ pytest.param(
+ BackfillPostBody,
+ "dag_run_conf",
+ {"dag_id": "dag1", "from_date": _DATE_A, "to_date": _DATE_B},
+ id="backfill",
+ ),
+]
+
+
[email protected](("model_cls", "field", "base_kwargs"), _CONF_BODIES)
+class TestConfReservedKeys:
+ """Dag run conf rejects reserved serialization keys the way XCom values
do."""
+
+ @pytest.mark.parametrize("reserved_key", sorted(FORBIDDEN_XCOM_KEYS))
+ def test_rejects_reserved_key_at_top_level(self, model_cls, field,
base_kwargs, reserved_key):
+ with pytest.raises(ValidationError, match="reserved serialization
keys") as ctx:
+ model_cls(**base_kwargs, **{field: {reserved_key: "x"}})
+ assert reserved_key in str(ctx.value)
+
+ @pytest.mark.parametrize(
+ "conf",
+ [
+ pytest.param(
+ {"a": {"__classname__": "builtins.dict", "__version__": 0,
"__data__": {}}},
+ id="current-nested",
+ ),
+ pytest.param({"a": [{"b": {"__type": "datetime", "__var": 0.0}}]},
id="legacy-nested-in-list"),
+ ],
+ )
+ def test_rejects_nested_reserved_keys(self, model_cls, field, base_kwargs,
conf):
+ with pytest.raises(ValidationError, match=rf"{field}\.a.* contains
reserved serialization keys"):
+ model_cls(**base_kwargs, **{field: conf})
+
+ @pytest.mark.parametrize(
+ "conf",
+ [
+ pytest.param(None, id="none"),
+ pytest.param({}, id="empty"),
+ pytest.param({"env": "dev", "nested": {"type": "x", "var": [1,
2]}}, id="plain"),
+ pytest.param({"a": json.dumps({"safe": "data"})},
id="benign-json-string"),
+ pytest.param({"a": json.dumps({"__type": "datetime", "__var":
0.0})}, id="legacy-in-json-string"),
+ pytest.param(
+ {"a": json.dumps({"b": {"__classname__": "builtins.dict"}})},
+ id="current-nested-in-json-string",
+ ),
+ pytest.param({"a": '{"not valid json'}, id="not-json"),
+ ],
+ )
+ def test_accepts_conf_without_reserved_keys(self, model_cls, field,
base_kwargs, conf):
+ assert getattr(model_cls(**base_kwargs, **{field: conf}), field) ==
conf
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
index 5cf95214368..310cc8470fd 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_assets.py
@@ -2522,6 +2522,21 @@ class TestPostAssetMaterialize(TestAssets):
assert response.json()["partition_key"] == "2026-03-23"
assert response.json()["run_type"] == "asset_materialization"
+ @pytest.mark.parametrize(
+ ("key", "conf"),
+ [
+ ("__classname__", {"__classname__": "builtins.dict",
"__version__": 0, "__data__": {}}),
+ ("__type", {"nested": {"__type": "datetime", "__var": 0.0}}),
+ ],
+ )
+ def test_should_respond_422_for_reserved_conf_keys(self, test_client,
session, key, conf):
+ response = test_client.post("/assets/1/materialize", json={"conf":
conf})
+ assert response.status_code == 422
+ detail = str(response.json()["detail"])
+ assert "reserved serialization keys" in detail
+ assert key in detail
+ assert session.scalar(select(func.count()).select_from(DagRun)) == 0
+
@pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
def test_should_respond_200_with_trigger_fields_without_dag_run_id(self,
test_client):
payload = {
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
index 8f2638f062b..5c852117c1d 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_backfills.py
@@ -1109,6 +1109,33 @@ class TestCreateBackfill(TestBackfillEndpoint):
response = unauthorized_test_client.post("/backfills", json=data)
assert response.status_code == 403
+ @pytest.mark.parametrize("url", ["/backfills", "/backfills/dry_run"])
+ @pytest.mark.parametrize(
+ ("key", "dag_run_conf"),
+ [
+ ("__classname__", {"__classname__": "builtins.dict",
"__version__": 0, "__data__": {}}),
+ ("__type", {"nested": {"__type": "datetime", "__var": 0.0}}),
+ ],
+ )
+ def test_create_backfill_rejects_reserved_conf_keys(
+ self, session, dag_maker, test_client, url, key, dag_run_conf
+ ):
+ with dag_maker(session=session, dag_id="TEST_DAG_CONF_KEYS",
schedule="0 * * * *") as dag:
+ EmptyOperator(task_id="mytask")
+ session.commit()
+ data = {
+ "dag_id": dag.dag_id,
+ "from_date": to_iso(pendulum.parse("2024-01-01")),
+ "to_date": to_iso(pendulum.parse("2024-01-02")),
+ "dag_run_conf": dag_run_conf,
+ }
+ response = test_client.post(url, json=data)
+ assert response.status_code == 422
+ detail = str(response.json()["detail"])
+ assert "reserved serialization keys" in detail
+ assert key in detail
+ assert session.scalar(select(func.count()).select_from(Backfill)) == 0
+
class TestCreateBackfillDryRun(TestBackfillEndpoint):
def test_create_backfill_dry_run_database_locked(self, session, dag_maker,
test_client):
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
index 68753886a13..36b9e4b6822 100644
--- a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
+++ b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_dag_run.py
@@ -17,6 +17,7 @@
from __future__ import annotations
+import json
import math
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
@@ -3659,6 +3660,30 @@ class TestTriggerDagRun:
assert response.status_code == 422
assert response.json() == expected_detail
+ @pytest.mark.parametrize(
+ ("key", "conf"),
+ [
+ ("__classname__", {"__classname__": "builtins.dict",
"__version__": 0, "__data__": {}}),
+ ("__type", {"nested": {"__type": "datetime", "__var": 0.0}}),
+ ],
+ )
+ def test_trigger_dag_run_rejects_reserved_conf_keys(self, test_client,
session, key, conf):
+ count_runs =
select(func.count()).select_from(DagRun).where(DagRun.dag_id == DAG1_ID)
+ runs_before = session.scalar(count_runs)
+ response = test_client.post(f"/dags/{DAG1_ID}/dagRuns", json={"conf":
conf, "logical_date": None})
+ assert response.status_code == 422
+ detail = str(response.json()["detail"])
+ assert "reserved serialization keys" in detail
+ assert key in detail
+ assert session.scalar(count_runs) == runs_before
+
+ @pytest.mark.usefixtures("configure_git_connection_for_dag_bundle")
+ def test_trigger_dag_run_keeps_json_string_conf_value(self, test_client):
+ conf = {"nested": json.dumps({"__type": "datetime", "__var": 0.0})}
+ response = test_client.post(f"/dags/{DAG1_ID}/dagRuns", json={"conf":
conf, "logical_date": None})
+ assert response.status_code == 200
+ assert response.json()["conf"] == conf
+
def test_post_dag_runs_with_empty_payload(self, test_client):
response = test_client.post(
f"/dags/{DAG1_ID}/dagRuns", data={}, headers={"Content-Type":
"application/json"}
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py
index cd78af871c5..87398769a5c 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_dag_runs.py
@@ -70,6 +70,27 @@ class TestDagRunTrigger:
assert dag_run.logical_date == logical_date
assert dag_run.run_type == DagRunType.OPERATOR_TRIGGERED
+ @pytest.mark.parametrize(
+ ("key", "conf"),
+ [
+ ("__classname__", {"__classname__": "builtins.dict",
"__version__": 0, "__data__": {}}),
+ ("__type", {"nested": [{"__type": "datetime", "__var": 0.0}]}),
+ ],
+ )
+ def test_trigger_dag_run_rejects_reserved_conf_keys(self, client, session,
dag_maker, key, conf):
+ dag_id = "test_trigger_dag_run_reserved_conf"
+ with dag_maker(dag_id=dag_id, session=session, serialized=True):
+ EmptyOperator(task_id="test_task")
+ session.commit()
+
+ response = client.post(f"/execution/dag-runs/{dag_id}/test_run_id",
json={"conf": conf})
+
+ assert response.status_code == 422
+ detail = str(response.json()["detail"])
+ assert "reserved serialization keys" in detail
+ assert key in detail
+ assert session.scalars(select(DagRun).where(DagRun.dag_id ==
dag_id)).all() == []
+
def test_trigger_dag_run_with_partition_key(self, client, session,
dag_maker):
"""partition_key accepted when Dag uses a partitioned timetable
(happy-path guard)."""
dag_id = "test_trigger_dag_run_partition_key"
diff --git a/airflow-core/tests/unit/jobs/test_scheduler_job.py
b/airflow-core/tests/unit/jobs/test_scheduler_job.py
index 8b575a8369a..9363b694775 100644
--- a/airflow-core/tests/unit/jobs/test_scheduler_job.py
+++ b/airflow-core/tests/unit/jobs/test_scheduler_job.py
@@ -11357,6 +11357,65 @@ class TestSchedulerJob:
# Assert that all deadlines which are both expired and unhandled get
processed.
assert mock_handle_miss.call_count == 2
+ @pytest.mark.parametrize(
+ "reserved_value",
+ [
+ pytest.param({"__type": "airflow.ConfProbe", "__var": {}},
id="legacy"),
+ pytest.param(
+ {"__classname__": "airflow.ConfProbe", "__version__": 0,
"__data__": {}}, id="current"
+ ),
+ ],
+ )
+ def
test_expired_deadline_with_reserved_keys_in_conf_does_not_stop_other_deadlines(
+ self, session, dag_maker, reserved_value
+ ):
+ """The scheduler keeps handling other deadlines and does not pick this
one up again."""
+ dag_id = "test_deadline_reserved_conf"
+ with dag_maker(dag_id=dag_id, session=session):
+ EmptyOperator(task_id="empty")
+ reserved_run = dag_maker.create_dagrun(run_id="reserved",
conf={"probe": reserved_value})
+ other_run = dag_maker.create_dagrun(
+ run_id="other", logical_date=timezone.utcnow() -
timedelta(days=1), conf={"probe": "plain"}
+ )
+ past_date = timezone.utcnow() - timedelta(minutes=5)
+ reserved_deadline = Deadline(
+ deadline_time=past_date,
+ callback=AsyncCallback("classpath.notify"),
+ dagrun_id=reserved_run.id,
+ dag_id=dag_id,
+ deadline_alert_id=None,
+ )
+ other_deadline = Deadline(
+ deadline_time=past_date,
+ callback=AsyncCallback("classpath.notify"),
+ dagrun_id=other_run.id,
+ dag_id=dag_id,
+ deadline_alert_id=None,
+ )
+ session.add_all([reserved_deadline, other_deadline])
+ session.commit()
+
+ with mock.patch.object(
+ Deadline, "handle_miss", autospec=True,
side_effect=Deadline.handle_miss
+ ) as spy:
+ for _ in range(2):
+ self.job_runner = SchedulerJobRunner(job=Job(), num_runs=1,
executors=[MockExecutor()])
+ self.job_runner._execute()
+
+ assert spy.call_count == 2
+ session.expire_all()
+ reserved_deadline = session.get(Deadline, reserved_deadline.id)
+ other_deadline = session.get(Deadline, other_deadline.id)
+ assert reserved_deadline.missed
+ assert reserved_deadline.callback.state == CallbackState.FAILED
+ assert other_deadline.missed
+ assert other_deadline.callback.state == CallbackState.QUEUED
+ assert [t.id for t in session.scalars(select(Trigger))] ==
[other_deadline.callback.trigger_id]
+
+ # The shared cleanup deletes triggers before callbacks.
+ session.execute(update(Callback).values(trigger_id=None))
+ session.commit()
+
@mock.patch("airflow.models.Deadline.handle_miss")
def test_process_expired_deadlines_no_deadlines_found(self,
mock_handle_miss, session):
"""Test handling when there are no deadlines to process."""
diff --git a/airflow-core/tests/unit/models/test_deadline.py
b/airflow-core/tests/unit/models/test_deadline.py
index 6c09ff069b1..fa009620f0b 100644
--- a/airflow-core/tests/unit/models/test_deadline.py
+++ b/airflow-core/tests/unit/models/test_deadline.py
@@ -30,8 +30,9 @@ from sqlalchemy import select
from airflow.api_fastapi.core_api.datamodels.dag_run import DAGRunResponse
from airflow.models import DagRun
from airflow.models.deadline import Deadline, ReferenceModels
+from airflow.models.trigger import Trigger
from airflow.providers.standard.operators.empty import EmptyOperator
-from airflow.sdk import timezone
+from airflow.sdk import serde, timezone
from airflow.sdk.definitions.callback import AsyncCallback, SyncCallback
from airflow.sdk.definitions.deadline import (
AverageRuntimeDeadline,
@@ -43,7 +44,7 @@ from airflow.sdk.definitions.deadline import (
deadline_reference,
)
from airflow.serialization.definitions.deadline import
SerializedReferenceModels
-from airflow.utils.state import DagRunState
+from airflow.utils.state import CallbackState, DagRunState
from tests_common.test_utils import db
from unit.models import DEFAULT_DATE
@@ -297,6 +298,58 @@ class TestDeadline:
callback.trigger = None
session.commit()
+ @pytest.mark.db_test
+ @pytest.mark.parametrize(
+ ("reserved_value", "keys"),
+ [
+ pytest.param({"__type": "airflow.ConfProbe", "__var": {}},
"__type, __var", id="legacy"),
+ pytest.param(
+ {"__classname__": "airflow.ConfProbe", "__version__": 0,
"__data__": {}},
+ "__classname__, __data__, __version__",
+ id="current",
+ ),
+ ],
+ )
+ @mock.patch("airflow.models.deadline.stats")
+ def
test_handle_miss_fails_triggerer_callback_with_reserved_keys_in_context(
+ self, mock_stats, dagrun, session, reserved_value, keys
+ ):
+ dagrun.conf = {"nested": {"probe": reserved_value}}
+ session.flush()
+ deadline_orm = Deadline(
+ deadline_time=DEFAULT_DATE,
+ callback=AsyncCallback(TEST_CALLBACK_PATH, TEST_CALLBACK_KWARGS),
+ dagrun_id=dagrun.id,
+ dag_id=dagrun.dag_id,
+ deadline_alert_id=None,
+ )
+ session.add(deadline_orm)
+ session.flush()
+
+ probe = mock.MagicMock()
+ real_import_string = serde.import_string
+
+ def import_string(path):
+ return probe if path == "airflow.ConfProbe" else
real_import_string(path)
+
+ with mock.patch.object(serde, "import_string",
side_effect=import_string):
+ deadline_orm.handle_miss(session)
+ session.flush()
+ for trigger in session.scalars(select(Trigger)):
+ Trigger._decrypt_kwargs(trigger.encrypted_kwargs)
+
+ probe.assert_not_called()
+ probe.deserialize.assert_not_called()
+ assert session.scalars(select(Trigger)).all() == []
+ callback = deadline_orm.callback
+ assert callback.trigger is None
+ assert callback.state == CallbackState.FAILED
+ assert callback.output == f"Not queued: the callback context holds
reserved serialization keys {keys}"
+ assert "context" not in callback.data["kwargs"]
+ assert deadline_orm.missed
+ assert session.scalars(select(Deadline).where(~Deadline.missed)).all()
== []
+
mock_stats.incr.assert_any_call(stat="deadline_alerts.callback_failed",
tags=mock.ANY)
+
@pytest.mark.db_test
def test_handle_miss_persists_executor_callback_routing_data(self, dagrun,
session):
deadline_orm = Deadline(
