Add async connection testing via workers for security isolation

.codespellignorelines

@@ -5,3 +5,4 @@
 <pre><code>Code block\ndoes not\nrespect\nnewlines\n</code></pre>
       "trough",
                 assert "task_instance_id" in route.dependant.path_param_names, (
+                assert "connection_test_id" in route.dependant.path_param_names, (

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                                                  |
 +=========================+==================+===================+==============================================================+
-| ``a7f3b2c1d4e5`` (head) | ``b8f3e4a1d2c9`` | ``3.3.0``         | Add allow_producer_teams column to                           |
+| ``a7e6d4c3b2f1`` (head) | ``a7f3b2c1d4e5`` | ``3.3.0``         | Add connection_test_request table for the deferred           |
+|                         |                  |                   | connection-test workflow.                                    |
++-------------------------+------------------+-------------------+--------------------------------------------------------------+
+| ``a7f3b2c1d4e5``        | ``b8f3e4a1d2c9`` | ``3.3.0``         | Add allow_producer_teams column to                           |
 |                         |                  |                   | dag_schedule_asset_reference table.                          |
 +-------------------------+------------------+-------------------+--------------------------------------------------------------+
 | ``b8f3e4a1d2c9``        | ``fde9ed84d07b`` | ``3.3.0``         | Add retry_delay_override and retry_reason to task_instance.  |

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

@@ -19,6 +19,7 @@
 
 import json
 from collections.abc import Iterable, Mapping
+from datetime import datetime
 from typing import Annotated, Any
 
 from pydantic import Field, field_validator, model_validator
@@ -78,12 +79,30 @@ class ConnectionCollectionResponse(BaseModel):
 
 
 class ConnectionTestResponse(BaseModel):
-    """Connection Test serializer for responses."""
+    """Connection Test serializer for synchronous test responses."""
 
     status: bool
     message: str
 
 
+class ConnectionTestQueuedResponse(BaseModel):
+    """Response returned when a connection test has been enqueued for worker execution."""
+
+    token: str
+    connection_id: str
+    state: str
+
+
+class AsyncConnectionTestResponse(BaseModel):
+    """Response returned when polling for the status of an enqueued connection test."""
+
+    token: str
+    connection_id: str
+    state: str
+    result_message: str | None = None
+    created_at: datetime
+
+
 class ConnectionHookFieldBehavior(BaseModel):
     """A class to store the behavior of each standard field of a Hook."""
 
@@ -210,3 +229,26 @@ def validate_team_name(self) -> ConnectionBody:
 
 
 ConnectionBodyPartial = make_partial_model(ConnectionBody)
+
+
+class ConnectionTestRequestBody(ConnectionBody):
+    """
+    Request body for enqueueing a connection test on a worker.
+
+    Inherits ``connection_id`` pattern, ``extra`` JSON validation, and
+    ``team_name`` handling from ``ConnectionBody`` so tested connections share
+    the same input contract as persisted ones.
+    """
+
+    commit_on_success: bool = Field(
+        default=False,
+        description="If True, save or update the connection in the connection table when the test succeeds.",
+    )
+    executor: str | None = Field(
+        default=None,
+        description="Executor name to dispatch the connection test to.",
+    )
+    queue: str | None = Field(
+        default=None,
+        description="Worker queue to route the connection test to (executor-dependent).",
+    )

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

@@ -1844,6 +1844,102 @@ paths:
       security:
       - OAuth2PasswordBearer: []
       - HTTPBearer: []
+  /api/v2/connections/enqueue-test:
+    post:
+      tags:
+      - Connection
+      summary: Enqueue Connection Test
+      description: Enqueue a connection test for deferred execution on a worker; returns
+        a polling token.
+      operationId: enqueue_connection_test
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/ConnectionTestRequestBody'
+        required: true
+      responses:
+        '202':
+          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'
+        '409':
+          description: Conflict
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+        '422':
+          description: Unprocessable Entity
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+  /api/v2/connections/enqueue-test/{connection_test_token}:
+    get:
+      tags:
+      - Connection
+      summary: Get Connection Test
+      description: Poll for the status of an enqueued connection test by its token.
+      operationId: get_connection_test
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: connection_test_token
+        in: path
+        required: true
+        schema:
+          type: string
+          title: Connection Test Token
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/AsyncConnectionTestResponse'
+        '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/defaults:
     post:
       tags:
@@ -10596,6 +10692,35 @@ components:
       - created_date
       title: AssetWatcherResponse
       description: Asset watcher serializer for responses.
+    AsyncConnectionTestResponse:
+      properties:
+        token:
+          type: string
+          title: Token
+        connection_id:
+          type: string
+          title: Connection Id
+        state:
+          type: string
+          title: State
+        result_message:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Result Message
+        created_at:
+          type: string
+          format: date-time
+          title: Created At
+      type: object
+      required:
+      - token
+      - connection_id
+      - state
+      - created_at
+      title: AsyncConnectionTestResponse
+      description: Response returned when polling for the status of an enqueued connection
+        test.
     BackfillCollectionResponse:
       properties:
         backfills:
@@ -11527,6 +11652,108 @@ components:
       - team_name
       title: ConnectionResponse
       description: Connection serializer for responses.
+    ConnectionTestQueuedResponse:
+      properties:
+        token:
+          type: string
+          title: Token
+        connection_id:
+          type: string
+          title: Connection Id
+        state:
+          type: string
+          title: State
+      type: object
+      required:
+      - token
+      - connection_id
+      - state
+      title: ConnectionTestQueuedResponse
+      description: Response returned when a connection test has been enqueued for
+        worker execution.
+    ConnectionTestRequestBody:
+      properties:
+        connection_id:
+          type: string
+          maxLength: 200
+          pattern: ^[\w.-]+$
+          title: Connection Id
+        conn_type:
+          type: string
+          title: Conn Type
+        description:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Description
+        host:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Host
+        login:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Login
+        schema:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Schema
+        port:
+          anyOf:
+          - type: integer
+          - type: 'null'
+          title: Port
+        password:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Password
+        extra:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Extra
+        team_name:
+          anyOf:
+          - type: string
+            maxLength: 50
+          - type: 'null'
+          title: Team Name
+        commit_on_success:
+          type: boolean
+          title: Commit On Success
+          description: If True, save or update the connection in the connection table
+            when the test succeeds.
+          default: false
+        executor:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Executor
+          description: Executor name to dispatch the connection test to.
+        queue:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Queue
+          description: Worker queue to route the connection test to (executor-dependent).
+      additionalProperties: false
+      type: object
+      required:
+      - connection_id
+      - conn_type
+      title: ConnectionTestRequestBody
+      description: 'Request body for enqueueing a connection test on a worker.
+
+
+        Inherits ``connection_id`` pattern, ``extra`` JSON validation, and
+
+        ``team_name`` handling from ``ConnectionBody`` so tested connections share
+
+        the same input contract as persisted ones.'
     ConnectionTestResponse:
       properties:
         status:
@@ -11540,7 +11767,7 @@ components:
       - status
       - message
       title: ConnectionTestResponse
-      description: Connection Test serializer for responses.
+      description: Connection Test serializer for synchronous test responses.
     CreateAssetEventsBody:
       properties:
         asset_id: