[WIP] Move connection testing to workers for security isolation

airflow-core/docs/img/airflow_erd.sha256

@@ -1 +1 @@
-174cc04e341bf41946034e5ba04dcc1da7b5b2c7e09cd74b378e07bb87de2ee6
+66d288a8d55d9d4b06e3c4ff10d5ebff298c8eb22756c7da1297fe5e55b805fe

airflow-core/docs/migrations-ref.rst

@@ -39,7 +39,10 @@ Here's the list of all the Database Migrations that are executed via when you ru
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | Revision ID             | Revises ID       | Airflow Version   | Description                                                  |
 +=========================+==================+===================+==============================================================+
-| ``e79fc784f145`` (head) | ``0b112f49112d`` | ``3.2.0``         | add timetable_type to dag table for filtering.               |
+| ``9882c124ea54`` (head) | ``e79fc784f145`` | ``3.2.0``         | Add connection_test_request table for async connection       |
+|                         |                  |                   | testing on workers.                                          |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``e79fc784f145``        | ``0b112f49112d`` | ``3.2.0``         | add timetable_type to dag table for filtering.               |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``0b112f49112d``        | ``c47f2e1ab9d4`` | ``3.2.0``         | Add exceeds max runs flag to dag model.                      |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+

airflow-core/src/airflow/api_fastapi/core_api/datamodels/connection_test.py

@@ -0,0 +1,44 @@
+# 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.
+
+"""Datamodels for async connection testing."""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from airflow.api_fastapi.core_api.base import BaseModel
+
+
+class ConnectionTestQueuedResponse(BaseModel):
+    """Response when a connection test is queued for async execution."""
+
+    request_id: str
+    state: str
+    message: str
+
+
+class ConnectionTestStatusResponse(BaseModel):
+    """Response with the full status of a connection test request."""
+
+    request_id: str
+    state: str
+    result_status: bool | None
+    result_message: str | None
+    created_at: datetime
+    started_at: datetime | None
+    completed_at: datetime | None

airflow-core/src/airflow/api_fastapi/core_api/openapi/v2-rest-api-generated.yaml

@@ -1297,6 +1297,112 @@ paths:
       security:
       - OAuth2PasswordBearer: []
       - HTTPBearer: []
+  /api/v2/connections/test:
+    post:
+      tags:
+      - Connection
+      summary: Test Connection
+      description: 'Queue a connection test for asynchronous execution on a worker.
+
+
+        This endpoint queues the connection test request for execution on a worker
+        node,
+
+        which provides better security isolation (workers run in ephemeral environments)
+
+        and network accessibility (workers can reach external systems that API servers
+        cannot).
+
+
+        Returns a request_id that can be used to poll for the test result via GET
+        /connections/test/{request_id}.'
+      operationId: test_connection
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/ConnectionBody'
+        required: true
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ConnectionTestQueuedResponse'
+        '401':
+          description: Unauthorized
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '403':
+          description: Forbidden
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+  /api/v2/connections/test/{request_id}:
+    get:
+      tags:
+      - Connection
+      summary: Get Connection Test Status
+      description: 'Get the status of a connection test request.
+
+
+        Poll this endpoint to check if a connection test has completed and retrieve
+        the result.'
+      operationId: get_connection_test_status
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: request_id
+        in: path
+        required: true
+        schema:
+          type: string
+          title: Request Id
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ConnectionTestStatusResponse'
+        '401':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Unauthorized
+        '403':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Forbidden
+        '404':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Not Found
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /api/v2/connections/{connection_id}:
     delete:
       tags:
@@ -1631,56 +1737,6 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
-  /api/v2/connections/test:
-    post:
-      tags:
-      - Connection
-      summary: Test Connection
-      description: 'Test an API connection.
-
-
-        This method first creates an in-memory transient conn_id & exports that to
-        an env var,
-
-        as some hook classes tries to find out the `conn` from their __init__ method
-        & errors out if not found.
-
-        It also deletes the conn id env connection after the test.'
-      operationId: test_connection
-      requestBody:
-        content:
-          application/json:
-            schema:
-              $ref: '#/components/schemas/ConnectionBody'
-        required: true
-      responses:
-        '200':
-          description: Successful Response
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/ConnectionTestResponse'
-        '401':
-          description: Unauthorized
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/HTTPExceptionResponse'
-        '403':
-          description: Forbidden
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/HTTPExceptionResponse'
-        '422':
-          description: Validation Error
-          content:
-            application/json:
-              schema:
-                $ref: '#/components/schemas/HTTPValidationError'
-      security:
-      - OAuth2PasswordBearer: []
-      - HTTPBearer: []
   /api/v2/connections/defaults:
     post:
       tags:
@@ -10066,20 +10122,69 @@ components:
       - team_name
       title: ConnectionResponse
       description: Connection serializer for responses.
-    ConnectionTestResponse:
+    ConnectionTestQueuedResponse:
       properties:
-        status:
-          type: boolean
-          title: Status
+        request_id:
+          type: string
+          title: Request Id
+        state:
+          type: string
+          title: State
         message:
           type: string
           title: Message
       type: object
       required:
-      - status
+      - request_id
+      - state
       - message
-      title: ConnectionTestResponse
-      description: Connection Test serializer for responses.
+      title: ConnectionTestQueuedResponse
+      description: Response when a connection test is queued for async execution.
+    ConnectionTestStatusResponse:
+      properties:
+        request_id:
+          type: string
+          title: Request Id
+        state:
+          type: string
+          title: State
+        result_status:
+          anyOf:
+          - type: boolean
+          - type: 'null'
+          title: Result Status
+        result_message:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Result Message
+        created_at:
+          type: string
+          format: date-time
+          title: Created At
+        started_at:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Started At
+        completed_at:
+          anyOf:
+          - type: string
+            format: date-time
+          - type: 'null'
+          title: Completed At
+      type: object
+      required:
+      - request_id
+      - state
+      - result_status
+      - result_message
+      - created_at
+      - started_at
+      - completed_at
+      title: ConnectionTestStatusResponse
+      description: Response with the full status of a connection test request.
     CreateAssetEventsBody:
       properties:
         asset_id: