This is an automated email from the ASF dual-hosted git repository.
shahar1 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 efc3844e981 Add tests for CoordinatorManager configuration error paths
(#69184)
efc3844e981 is described below
commit efc3844e98150c6c1ee52ce2e76b8028c777cfa9
Author: Andrew Chang <[email protected]>
AuthorDate: Tue Jun 30 20:43:51 2026 +0000
Add tests for CoordinatorManager configuration error paths (#69184)
The error branches that translate Python-internal exceptions
(KeyError, ImportError, TypeError, ValueError) into user-facing
InvalidCoordinatorError / ValueError messages in CoordinatorManager
had no test coverage. Without these tests, a refactor could
silently drop one of the except handlers and users would see raw
Python tracebacks instead of the friendly domain errors the
coordinator layer promises.
---
.../task_sdk/execution_time/test_coordinator.py | 52 ++++++++++++++++++++++
1 file changed, 52 insertions(+)
diff --git a/task-sdk/tests/task_sdk/execution_time/test_coordinator.py
b/task-sdk/tests/task_sdk/execution_time/test_coordinator.py
index 40db5378afc..73e737bfb8b 100644
--- a/task-sdk/tests/task_sdk/execution_time/test_coordinator.py
+++ b/task-sdk/tests/task_sdk/execution_time/test_coordinator.py
@@ -27,6 +27,7 @@ from airflow.sdk.configuration import conf,
retrieve_configuration_description
from airflow.sdk.execution_time.coordinator import (
BaseCoordinator,
CoordinatorManager,
+ InvalidCoordinatorError,
_PythonCoordinator,
get_coordinator_manager,
reset_coordinator_manager,
@@ -116,6 +117,57 @@ class TestCoordinatorManager:
assert manager._coordinator_specs == {}
assert manager._queue_to_coordinator == {}
+ def test_from_config_rejects_invalid_queue_mapping(self, sdk_config):
+ sdk_config(
+ coordinators=json.dumps(
+ {"alpha": {"classpath":
f"{_CoordinatorA.__module__}._CoordinatorA", "kwargs": {}}}
+ ),
+ queue_to_coordinator=json.dumps({"queue-x": "nonexistent"}),
+ )
+ with pytest.raises(
+ ValueError,
+ match=r"queue_to_coordinator references invalid coordinator key:
'nonexistent'",
+ ):
+ CoordinatorManager.from_config()
+
+ @pytest.mark.parametrize(
+ ("coordinator_spec", "expected_match"),
+ [
+ pytest.param(
+ {"classpath": "nonexistent.module.MissingCoordinator",
"kwargs": {}},
+ r"Cannot import coordinator 'bad-coord'",
+ id="classpath-cannot-be-imported",
+ ),
+ pytest.param(
+ {
+ "classpath": f"{_CoordinatorA.__module__}._CoordinatorA",
+ "kwargs": {"unknown_param": "x"},
+ },
+ r"Cannot instantiate coordinator 'bad-coord'",
+ id="kwargs-dont-match-constructor",
+ ),
+ ],
+ )
+ def test_for_queue_raises_invalid_coordinator_error(self, sdk_config,
coordinator_spec, expected_match):
+ sdk_config(
+ coordinators=json.dumps({"bad-coord": coordinator_spec}),
+ queue_to_coordinator=json.dumps({"queue-bad": "bad-coord"}),
+ )
+ manager = CoordinatorManager.from_config()
+ with pytest.raises(InvalidCoordinatorError, match=expected_match):
+ manager.for_queue("queue-bad")
+
+ def test_for_queue_raises_when_spec_missing(self):
+ manager = CoordinatorManager(
+ coordinator_specs={},
+ queue_to_coordinator={"queue-bad": "missing"},
+ )
+ with pytest.raises(
+ InvalidCoordinatorError,
+ match=r"Queue 'queue-bad' configured to nonexistent coordinator",
+ ):
+ manager.for_queue("queue-bad")
+
def test_get_coordinator_manager_is_cached(self, monkeypatch):
monkeypatch.delenv("AIRFLOW__SDK__COORDINATORS", raising=False)
