[WIP] Move connection testing to workers for security isolation

[anishgirianish]

---

Summary

Demo:

Screencast.from.01-15-2026.09.56.31.PM.webm

This PR implements async connection testing on workers as recommended by @potiuk in #59643. Instead of running test_connection on the API server (which has security and network accessibility concerns), this moves the execution to workers where:

• Workers run in ephemeral environments (better security isolation)
• Workers have network access to external systems that API servers may not have
• This aligns with Airflow 3.2's direction of separating API servers from task execution

Architecture

UI/API → Queue request (DB: PENDING) → Scheduler dispatches → Worker executes → Reports result

1. POST /connections/test - Queues a ConnectionTestRequest with encrypted connection URI
2. Scheduler - Picks up PENDING requests, marks RUNNING, dispatches to executor
3. Worker - Decrypts URI, runs test_connection(), reports result via execution API
4. GET /connections/test/{request_id} - Poll for result

Key Implementation Details

• Uses Connection.get_connection_from_secrets() per maintainer feedback
• Uses update_orm_from_pydantic() to handle masked passwords ("***")
• Connection URI is Fernet-encrypted for secure storage/transport
• Single execution model (scheduler-push only, no worker-pull)
• Respects existing [core] test_connection config setting

Files Changed

• New model: ConnectionTestRequest with state machine (PENDING → RUNNING → SUCCESS/FAILED)
• New migration: Adds connection_test_request table
• Core API: /connections/test endpoints for queuing and polling
• Execution API: /{request_id}/state endpoint for workers to report results
• Scheduler: _dispatch_connection_tests() in heartbeat loop
• LocalExecutor: Handles TestConnection workload type

Testing

• Unit tests for model, core API routes, and execution API routes
• Manual testing with LocalExecutor

Questions for Maintainers

1. Is the scheduler the right place to dispatch connection tests, or should this use a different mechanism?
2. Should we add cleanup for stuck RUNNING requests (e.g., timeout after N minutes)?
3. Is there anything specific about the deadline alerts pattern we should follow more closely?

Looking forward to your feedback on the overall approach before polishing further.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.