This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 4c9f8f025af [v3-4-test] Keep queued connection tests from overwriting
newer connections (#74837) (#74895)
4c9f8f025af is described below
commit 4c9f8f025afa058092062ca53b0ce1a1ac1c19dd
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sat Oct 17 00:30:05 2026 +0100
[v3-4-test] Keep queued connection tests from overwriting newer connections
(#74837) (#74895)
(cherry picked from commit 3ce8d152d0dd5424512cbb58a7a756586f79e51e)
---
airflow-core/docs/migrations-ref.rst | 4 +-
airflow-core/newsfragments/74837.bugfix.rst | 1 +
.../core_api/routes/public/connections.py | 4 +-
.../execution_api/routes/connection_tests.py | 7 ++-
..._add_commit_allows_update_to_connection_test.py | 51 ++++++++++++++++++++++
airflow-core/src/airflow/models/connection_test.py | 26 ++++++++++-
airflow-core/src/airflow/utils/db.py | 2 +-
.../core_api/routes/public/test_connections.py | 21 +++++++++
.../versions/head/test_connection_tests.py | 33 ++++++++++++++
.../tests/unit/models/test_connection_test.py | 32 +++++++++++++-
10 files changed, 174 insertions(+), 7 deletions(-)
diff --git a/airflow-core/docs/migrations-ref.rst
b/airflow-core/docs/migrations-ref.rst
index 35315b7d044..44e2090370d 100644
--- a/airflow-core/docs/migrations-ref.rst
+++ b/airflow-core/docs/migrations-ref.rst
@@ -39,7 +39,9 @@ Here's the list of all the Database Migrations that are
executed via when you ru
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| Revision ID | Revises ID | Airflow Version | Description
|
+=========================+==================+===================+==============================================================+
-| ``a5d7b9c13e40`` (head) | ``c3e7a9182f64`` | ``3.4.0`` | Preserve
exact task execution attribution in asset state. |
+| ``9cffe90fa0c0`` (head) | ``a5d7b9c13e40`` | ``3.4.0`` | Add
commit_allows_update to connection_test_request. |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``a5d7b9c13e40`` | ``c3e7a9182f64`` | ``3.4.0`` | Preserve
exact task execution attribution in asset state. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
| ``c3e7a9182f64`` | ``7f8c9a2d410e`` | ``3.4.0`` | Attribute
log and asset events to exact task tries. |
+-------------------------+------------------+-------------------+--------------------------------------------------------------+
diff --git a/airflow-core/newsfragments/74837.bugfix.rst
b/airflow-core/newsfragments/74837.bugfix.rst
new file mode 100644
index 00000000000..7c974ce06c9
--- /dev/null
+++ b/airflow-core/newsfragments/74837.bugfix.rst
@@ -0,0 +1 @@
+A queued connection test with ``commit_on_success`` for a connection that did
not exist when the test was queued no longer overwrites a connection with the
same ID created in the meantime. The test is reported as failed with a message
explaining that the tested values were not saved.
diff --git
a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py
b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py
index d803b85b191..7fa0151bae8 100644
--- a/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py
+++ b/airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py
@@ -452,7 +452,8 @@ def enqueue_connection_test(
else:
effective_team = test_body.team_name
- auth_method: ResourceMethod = "PUT" if existing is not None and
test_body.commit_on_success else "POST"
+ commit_allows_update = existing is not None and test_body.commit_on_success
+ auth_method: ResourceMethod = "PUT" if commit_allows_update else "POST"
if not get_auth_manager().is_authorized_connection(
method=auth_method,
details=ConnectionDetails(conn_id=test_body.connection_id,
team_name=effective_team),
@@ -473,6 +474,7 @@ def enqueue_connection_test(
port=test_body.port,
extra=test_body.extra,
commit_on_success=test_body.commit_on_success,
+ commit_allows_update=commit_allows_update,
executor=test_body.executor,
queue=test_body.queue,
team_name=effective_team,
diff --git
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/connection_tests.py
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/connection_tests.py
index 21654109478..892243e182e 100644
---
a/airflow-core/src/airflow/api_fastapi/execution_api/routes/connection_tests.py
+++
b/airflow-core/src/airflow/api_fastapi/execution_api/routes/connection_tests.py
@@ -137,4 +137,9 @@ def patch_connection_test(
ct.result_message = body.result_message
if body.state == ConnectionTestState.SUCCESS and ct.commit_on_success:
- ct.commit_to_connection_table(session=session)
+ if not ct.commit_to_connection_table(session=session):
+ ct.state = ConnectionTestState.FAILED
+ ct.result_message = (
+ f"Connection test succeeded, but connection
`{ct.connection_id}` was created by "
+ "another request after this test was queued, so the tested
values were not saved."
+ )
diff --git
a/airflow-core/src/airflow/migrations/versions/0147_3_4_0_add_commit_allows_update_to_connection_test.py
b/airflow-core/src/airflow/migrations/versions/0147_3_4_0_add_commit_allows_update_to_connection_test.py
new file mode 100644
index 00000000000..16ac18f9703
--- /dev/null
+++
b/airflow-core/src/airflow/migrations/versions/0147_3_4_0_add_commit_allows_update_to_connection_test.py
@@ -0,0 +1,51 @@
+#
+# 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.
+
+"""
+Add commit_allows_update to connection_test_request.
+
+Revision ID: 9cffe90fa0c0
+Revises: a5d7b9c13e40
+Create Date: 2026-10-16 10:00:00.000000
+
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "9cffe90fa0c0"
+down_revision = "a5d7b9c13e40"
+branch_labels = None
+depends_on = None
+airflow_version = "3.4.0"
+
+
+def upgrade():
+ """Add the commit_allows_update column to connection_test_request."""
+ with op.batch_alter_table("connection_test_request", schema=None) as
batch_op:
+ batch_op.add_column(
+ sa.Column("commit_allows_update", sa.Boolean(), nullable=False,
server_default="0")
+ )
+
+
+def downgrade():
+ """Remove the commit_allows_update column from connection_test_request."""
+ with op.batch_alter_table("connection_test_request", schema=None) as
batch_op:
+ batch_op.drop_column("commit_allows_update")
diff --git a/airflow-core/src/airflow/models/connection_test.py
b/airflow-core/src/airflow/models/connection_test.py
index f3d79a2a34c..e9c0a3ec9b9 100644
--- a/airflow-core/src/airflow/models/connection_test.py
+++ b/airflow-core/src/airflow/models/connection_test.py
@@ -110,6 +110,12 @@ class ConnectionTestRequest(Base, FernetFieldsMixin):
commit_on_success: Mapped[bool] = mapped_column(
Boolean, nullable=False, default=False, server_default="0"
)
+ # Whether a successful commit may overwrite an existing connection. Set at
enqueue time only
+ # when the connection already existed and the requester was authorized to
edit it; otherwise
+ # the commit may only create the connection.
+ commit_allows_update: Mapped[bool] = mapped_column(
+ Boolean, nullable=False, default=False, server_default="0"
+ )
is_encrypted: Mapped[bool] = mapped_column(
Boolean, unique=False, default=False, nullable=False,
server_default="0"
)
@@ -140,6 +146,7 @@ class ConnectionTestRequest(Base, FernetFieldsMixin):
port: int | None = None,
extra: str | None = None,
commit_on_success: bool = False,
+ commit_allows_update: bool = False,
executor: str | None = None,
queue: str | None = None,
team_name: str | None = None,
@@ -155,6 +162,7 @@ class ConnectionTestRequest(Base, FernetFieldsMixin):
self.port = port
self.extra = extra
self.commit_on_success = commit_on_success
+ self.commit_allows_update = commit_allows_update
self.executor = executor
self.queue = queue
self.team_name = team_name
@@ -202,8 +210,15 @@ class ConnectionTestRequest(Base, FernetFieldsMixin):
extra=self.extra,
)
- def commit_to_connection_table(self, *, session: Session) -> None:
- """Upsert the tested connection into the real ``connection`` table."""
+ def commit_to_connection_table(self, *, session: Session) -> bool:
+ """
+ Write the tested connection into the real ``connection`` table.
+
+ An existing connection is only overwritten when
``commit_allows_update`` is set, i.e. the
+ connection existed when the test was enqueued and the requester was
allowed to edit it.
+
+ :return: True if the connection was created or updated, False if it
was left unchanged.
+ """
conn =
session.scalar(select(Connection).filter_by(conn_id=self.connection_id))
if conn is None:
conn = Connection(
@@ -219,6 +234,12 @@ class ConnectionTestRequest(Base, FernetFieldsMixin):
)
session.add(conn)
log.info("Created new connection from successful test",
connection_id=self.connection_id)
+ elif not self.commit_allows_update:
+ log.warning(
+ "Not updating connection created after the test was enqueued",
+ connection_id=self.connection_id,
+ )
+ return False
else:
conn.conn_type = self.conn_type
conn.host = self.host
@@ -228,3 +249,4 @@ class ConnectionTestRequest(Base, FernetFieldsMixin):
conn.port = self.port
conn.extra = self.extra
log.info("Updated existing connection from successful test",
connection_id=self.connection_id)
+ return True
diff --git a/airflow-core/src/airflow/utils/db.py
b/airflow-core/src/airflow/utils/db.py
index 594bfae64c4..5f0f2d245cc 100644
--- a/airflow-core/src/airflow/utils/db.py
+++ b/airflow-core/src/airflow/utils/db.py
@@ -117,7 +117,7 @@ _REVISION_HEADS_MAP: dict[str, str] = {
"3.1.8": "509b94a1042d",
"3.2.0": "1d6611b6ab7c",
"3.3.0": "d2f4e1b3c5a7",
- "3.4.0": "a5d7b9c13e40",
+ "3.4.0": "9cffe90fa0c0",
}
# Prefix used to identify tables holding data moved during migration.
diff --git
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
index 67c276b7300..f8735b498e1 100644
---
a/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
+++
b/airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py
@@ -1610,6 +1610,27 @@ class TestAsyncConnectionTest(TestConnectionEndpoint):
assert ct is not None
assert ct.commit_on_success is True
+ @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+ def test_post_commit_on_success_new_connection_is_create_only(self,
test_client, session):
+ """A commit_on_success test for a connection that does not exist yet
may only create it."""
+ body = {**self.TEST_REQUEST_BODY, "commit_on_success": True}
+ response = test_client.post("/connections/enqueue-test", json=body)
+ assert response.status_code == 202
+
+ ct =
session.scalar(select(ConnectionTestRequest).filter_by(token=response.json()["token"]))
+ assert ct.commit_allows_update is False
+
+ @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+ def test_post_commit_on_success_existing_connection_allows_update(self,
test_client, session):
+ """A commit_on_success test for an existing connection, authorized for
edit, may update it."""
+ self.create_connection()
+ body = {**self.TEST_REQUEST_BODY, "commit_on_success": True}
+ response = test_client.post("/connections/enqueue-test", json=body)
+ assert response.status_code == 202
+
+ ct =
session.scalar(select(ConnectionTestRequest).filter_by(token=response.json()["token"]))
+ assert ct.commit_allows_update is True
+
@mock.patch("airflow.api_fastapi.core_api.routes.public.connections.get_auth_manager",
autospec=True)
@mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
def
test_post_commit_on_success_existing_connection_requires_edit_permission(
diff --git
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_connection_tests.py
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_connection_tests.py
index 6c699176589..a739cbcc62e 100644
---
a/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_connection_tests.py
+++
b/airflow-core/tests/unit/api_fastapi/execution_api/versions/head/test_connection_tests.py
@@ -140,6 +140,7 @@ class TestPatchConnectionTestCommitOnSuccess:
host="new-host.example.com",
login="new_user",
commit_on_success=True,
+ commit_allows_update=True,
)
ct.state = ConnectionTestState.RUNNING
session.add(ct)
@@ -156,6 +157,38 @@ class TestPatchConnectionTestCommitOnSuccess:
assert conn.conn_type == "postgres"
assert conn.host == "new-host.example.com"
+ def test_success_with_create_only_commit_does_not_overwrite_existing(self,
client, session):
+ """A create-only commit does not overwrite a connection created after
the test was enqueued."""
+ ct = ConnectionTestRequest(
+ connection_id="raced_conn",
+ conn_type="postgres",
+ host="tested-host.example.com",
+ login="tested_user",
+ password="tested_secret",
+ commit_on_success=True,
+ )
+ ct.state = ConnectionTestState.RUNNING
+ session.add(ct)
+ session.add(Connection(conn_id="raced_conn", conn_type="http",
host="other-host.example.com"))
+ session.commit()
+
+ response = client.patch(
+ f"/execution/connection-tests/{ct.id}",
+ json={"state": "success", "result_message": "Connection OK"},
+ )
+ assert response.status_code == 204
+
+ session.expire_all()
+ conn =
session.scalar(select(Connection).filter_by(conn_id="raced_conn"))
+ assert conn.conn_type == "http"
+ assert conn.host == "other-host.example.com"
+ assert conn.login is None
+ assert conn.password is None
+
+ ct = session.get(ConnectionTestRequest, ct.id)
+ assert ct.state == ConnectionTestState.FAILED
+ assert "were not saved" in ct.result_message
+
def test_success_without_commit_does_not_create(self, client, session):
"""PATCH with state=success but commit_on_success=False does not
create a connection."""
ct = ConnectionTestRequest(
diff --git a/airflow-core/tests/unit/models/test_connection_test.py
b/airflow-core/tests/unit/models/test_connection_test.py
index 1c27ac998a8..3cce66c3c01 100644
--- a/airflow-core/tests/unit/models/test_connection_test.py
+++ b/airflow-core/tests/unit/models/test_connection_test.py
@@ -261,11 +261,12 @@ class TestCommitToConnectionTable:
host="new-host.example.com",
login="new_user",
password="new_secret",
+ commit_allows_update=True,
)
session.add(ct)
session.flush()
- ct.commit_to_connection_table(session=session)
+ assert ct.commit_to_connection_table(session=session) is True
session.flush()
session.refresh(conn)
@@ -273,3 +274,32 @@ class TestCommitToConnectionTable:
assert conn.host == "new-host.example.com"
assert conn.login == "new_user"
assert conn.password == "new_secret"
+
+ def
test_does_not_update_existing_connection_without_commit_allows_update(self,
session):
+ """A create-only commit leaves a connection that appeared after
enqueue untouched."""
+ conn = Connection(conn_id="existing_conn", conn_type="http",
host="old-host.example.com")
+ session.add(conn)
+ session.flush()
+
+ ct = ConnectionTestRequest(
+ connection_id="existing_conn",
+ conn_type="postgres",
+ host="new-host.example.com",
+ login="new_user",
+ password="new_secret",
+ )
+ session.add(ct)
+ session.flush()
+
+ assert ct.commit_to_connection_table(session=session) is False
+ session.flush()
+ session.refresh(conn)
+
+ assert conn.conn_type == "http"
+ assert conn.host == "old-host.example.com"
+ assert conn.login is None
+ assert conn.password is None
+
+ def test_commit_allows_update_defaults_to_false(self):
+ ct = ConnectionTestRequest(connection_id="test_conn",
conn_type="postgres", commit_on_success=True)
+ assert ct.commit_allows_update is False
