This is an automated email from the ASF dual-hosted git repository.
potiuk pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 1298b23136e Add CeleryExecutor callback workload routing test (#66435)
1298b23136e is described below
commit 1298b23136e042337ff534f30b67519eb410253f
Author: Paul <[email protected]>
AuthorDate: Sat May 9 19:35:05 2026 -0700
Add CeleryExecutor callback workload routing test (#66435)
---
.../unit/celery/executors/test_celery_executor.py | 37 ++++++++++++++++++++++
1 file changed, 37 insertions(+)
diff --git
a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
index 309eba50cbd..3dcd32e84ef 100644
--- a/providers/celery/tests/unit/celery/executors/test_celery_executor.py
+++ b/providers/celery/tests/unit/celery/executors/test_celery_executor.py
@@ -885,6 +885,43 @@ def test_celery_tasks_registered_on_import():
)
[email protected](not AIRFLOW_V_3_2_PLUS, reason="ExecuteCallback requires
Airflow 3.2+")
[email protected](
+ ("callback_data", "expected_queue"),
+ [
+ pytest.param({"path": "tests.callbacks.test_callback", "kwargs": {}},
"default", id="default_queue"),
+ pytest.param(
+ {"path": "tests.callbacks.test_callback", "kwargs": {}, "queue":
"callback_queue"},
+ "callback_queue",
+ id="callback_queue",
+ ),
+ ],
+)
[email protected]("airflow.providers.celery.executors.celery_executor.CeleryExecutor._send_workloads")
+def test_process_workloads_routes_execute_callback(mock_send_workloads,
callback_data, expected_queue):
+ """CeleryExecutor routes callback workloads to Celery with the expected
queue."""
+ from airflow.executors import workloads
+ from airflow.executors.workloads.callback import CallbackDTO
+
+ callback_id = "00000000-0000-0000-0000-000000000003"
+ workload = workloads.ExecuteCallback(
+ callback=CallbackDTO(
+ id=callback_id,
+ fetch_method=workloads.CallbackFetchMethod.IMPORT_PATH,
+ data=callback_data,
+ ),
+ dag_rel_path="test_dag.py",
+ bundle_info=workloads.BundleInfo(name="test-bundle", version=None),
+ token="test-token",
+ log_path="callback.log",
+ )
+
+ executor = celery_executor.CeleryExecutor()
+ executor._process_workloads([workload])
+
+ mock_send_workloads.assert_called_once_with([(callback_id, workload,
expected_queue, None)])
+
+
@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="execute_workload is only
used for Airflow 3+")
@pytest.mark.skipif(AIRFLOW_V_3_3_PLUS, reason="pre-3.3 compatibility path
only applies before Airflow 3.3")
def test_execute_workload_runs_execute_task_before_airflow_3_3():
