[API] /connections/test: Use stored password for saved connections and fix DBT hook URL

airflow-core/src/airflow/api_fastapi/core_api/routes/public/connections.py

@@ -214,6 +214,7 @@ def patch_connection(
 @connections_router.post("/test", dependencies=[Depends(requires_access_connection(method="POST"))])
 def test_connection(
     test_body: ConnectionBody,
+    session: SessionDep,
 ) -> ConnectionTestResponse:
     """
     Test an API connection.
@@ -229,17 +230,27 @@ def test_connection(
             "Contact your deployment admin to enable it.",
         )
 
-    transient_conn_id = get_random_string()
-    conn_env_var = f"{CONN_ENV_PREFIX}{transient_conn_id.upper()}"
+    conn_env_var: str | None = None
     try:
-        data = test_body.model_dump(by_alias=True)
-        data["conn_id"] = transient_conn_id
-        conn = Connection(**data)
+        conn = None
+        if test_body.connection_id:
+            conn = session.scalar(select(Connection).filter_by(conn_id=test_body.connection_id))
+
+        if conn is None:
+            effective_conn_id = get_random_string()
+            data = test_body.model_dump(by_alias=True)
+            data["conn_id"] = effective_conn_id
+            conn = Connection(**data)
+        else:
+            effective_conn_id = conn.conn_id
+
+        conn_env_var = f"{CONN_ENV_PREFIX}{effective_conn_id.upper()}"
         os.environ[conn_env_var] = conn.get_uri()
         test_status, test_message = conn.test_connection()
         return ConnectionTestResponse.model_validate({"status": test_status, "message": test_message})
     finally:
-        os.environ.pop(conn_env_var, None)
+        if conn_env_var:
+            os.environ.pop(conn_env_var, None)
 
 
 @connections_router.post(

airflow-core/tests/unit/api_fastapi/core_api/routes/public/test_connections.py

@@ -945,21 +945,56 @@ def setup_method(self):
     @pytest.mark.parametrize(
         ("body", "message"),
         [
-            ({"connection_id": TEST_CONN_ID, "conn_type": "sqlite"}, "Connection successfully tested"),
             (
-                {"connection_id": TEST_CONN_ID, "conn_type": "fs", "extra": '{"path": "/"}'},
+                {
+                    "connection_id": TEST_CONN_ID,
+                    "conn_type": "fs",
+                    "extra": '{"path": "/"}',
+                    "password": "***",
+                },
                 "Path / is existing.",
             ),
         ],
     )
-    def test_should_respond_200(self, test_client, body, message):
+    def test_should_respond_200_saved_connection(self, test_client, session, body, message):
+        connection = Connection(
+            conn_id=TEST_CONN_ID,
+            conn_type="fs",
+            login="a",
+            password="a",
+            extra='{"path": "/"}',
+        )
+        session.add(connection)
+        session.commit()
+
         response = test_client.post("/connections/test", json=body)
         assert response.status_code == 200
         assert response.json() == {
             "status": True,
             "message": message,
         }
 
+    @mock.patch.dict(os.environ, {"AIRFLOW__CORE__TEST_CONNECTION": "Enabled"})
+    @pytest.mark.parametrize(
+        ("body", "message"),
+        [
+            (
+                {
+                    "connection_id": "pre_save_conn",
+                    "conn_type": "sqlite",
+                    "host": "example.com",
+                    "login": "user",
+                    "password": "pass",
+                },
+                "Connection successfully tested",
+            ),
+        ],
+    )
+    def test_should_respond_200_transient_connection(self, test_client, body, message):
+        response = test_client.post("/connections/test", json=body)
+        assert response.status_code == 200
+        assert response.json() == {"status": True, "message": message}
+
     def test_should_respond_401(self, unauthenticated_test_client):
         response = unauthenticated_test_client.post(
             "/connections/test", json={"connection_id": TEST_CONN_ID, "conn_type": "sqlite"}

providers/dbt/cloud/src/airflow/providers/dbt/cloud/hooks/dbt.py

@@ -926,10 +926,11 @@ def retry_failed_job_run(self, job_id: int, account_id: int | None = None) -> Re
         """
         return self._run_and_get_response(method="POST", endpoint=f"{account_id}/jobs/{job_id}/rerun/")
 
-    def test_connection(self) -> tuple[bool, str]:
+    @fallback_to_default_account
+    def test_connection(self, account_id: int | None = None) -> tuple[bool, str]:
         """Test dbt Cloud connection."""
         try:
-            self._run_and_get_response()
+            self._run_and_get_response(endpoint=f"{account_id}/")
             return True, "Successfully connected to dbt Cloud."
         except Exception as e:
             return False, str(e)

providers/dbt/cloud/tests/unit/dbt/cloud/hooks/test_dbt.py

@@ -1058,25 +1058,33 @@ def test_get_job_run_artifact_with_payload(self, mock_http_run, mock_paginate, c
         hook._paginate.assert_not_called()
 
     @pytest.mark.parametrize(
-        argnames="conn_id",
-        argvalues=[ACCOUNT_ID_CONN, NO_ACCOUNT_ID_CONN],
+        argnames=("conn_id", "account_id"),
+        argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],
         ids=["default_account", "explicit_account"],
     )
-    def test_connection_success(self, requests_mock, conn_id):
-        requests_mock.get(BASE_URL, status_code=200)
-        status, msg = DbtCloudHook(conn_id).test_connection()
+    @patch.object(
+        DbtCloudHook,
+        "run_with_advanced_retry",
+        return_value={"data": "success"},
+    )
+    def test_connection_success(self, _, conn_id, account_id):
+        status, msg = DbtCloudHook(conn_id).test_connection(account_id=account_id)
 
         assert status is True
         assert msg == "Successfully connected to dbt Cloud."
 
     @pytest.mark.parametrize(
-        argnames="conn_id",
-        argvalues=[ACCOUNT_ID_CONN, NO_ACCOUNT_ID_CONN],
+        argnames=("conn_id", "account_id"),
+        argvalues=[(ACCOUNT_ID_CONN, None), (NO_ACCOUNT_ID_CONN, ACCOUNT_ID)],
         ids=["default_account", "explicit_account"],
     )
-    def test_connection_failure(self, requests_mock, conn_id):
-        requests_mock.get(BASE_URL, status_code=403, reason="Authentication credentials were not provided")
-        status, msg = DbtCloudHook(conn_id).test_connection()
+    @patch.object(
+        DbtCloudHook,
+        "run_with_advanced_retry",
+        side_effect=Exception("403:Authentication credentials were not provided"),
+    )
+    def test_connection_failure(self, _, conn_id, account_id):
+        status, msg = DbtCloudHook(conn_id).test_connection(account_id=account_id)
 
         assert status is False
         assert msg == "403:Authentication credentials were not provided"