This is an automated email from the ASF dual-hosted git repository.
dheerajturaga 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 283d5c38d5c edge3: Exit non-zero when worker admin CLI commands fail
(#70112)
283d5c38d5c is described below
commit 283d5c38d5cca56ab662ac602927a269878418d5
Author: Deepak kumar <[email protected]>
AuthorDate: Sun Jul 26 10:20:58 2026 -0700
edge3: Exit non-zero when worker admin CLI commands fail (#70112)
The add-worker-queues, remove-worker-queues, set-worker-concurrency,
update-maintenance-comments, and remove-worker subcommands raised bare
`SystemExit` after catching a TypeError from the model layer, which
yields exit status 0. Shell scripts wrapping these commands treated
failures as success and continued as if the queue, concurrency,
maintenance-comment, or worker-removal change had been applied.
Pass the exception message to SystemExit so the process exits with
status 1 and writes the failure to stderr, matching the file's own
convention for validation-error paths (see the explicit
`raise SystemExit("Error: ...")` calls in the same subcommands) and the
`raise SystemExit(str(e))` pattern already used elsewhere in the edge3
CLI. The model layer already logs the error before raising, so the
redundant logger.error call in the three subcommands that had one is
removed.
---
.../airflow/providers/edge3/cli/edge_command.py | 17 ++---
.../tests/unit/edge3/cli/test_edge_command.py | 77 +++++++++++++++++++++-
2 files changed, 83 insertions(+), 11 deletions(-)
diff --git a/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
b/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
index 2088bd3bb75..7daf1b58a69 100644
--- a/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
+++ b/providers/edge3/src/airflow/providers/edge3/cli/edge_command.py
@@ -328,8 +328,8 @@ def remote_worker_update_maintenance_comment(args) -> None:
try:
change_maintenance_comment(args.edge_hostname, args.comments)
logger.info("Maintenance comments updated for %s by %s.",
args.edge_hostname, getuser())
- except TypeError:
- raise SystemExit
+ except TypeError as e:
+ raise SystemExit(str(e))
@cli_utils.action_cli(check_db=False)
@@ -343,8 +343,8 @@ def remove_remote_worker(args) -> None:
try:
remove_worker(args.edge_hostname)
logger.info("Edge Worker host %s removed by %s.", args.edge_hostname,
getuser())
- except TypeError:
- raise SystemExit
+ except TypeError as e:
+ raise SystemExit(str(e))
@cli_utils.action_cli(check_db=False)
@@ -406,8 +406,7 @@ def add_worker_queues(args) -> None:
add_worker_queues(args.edge_hostname, queues)
logger.info("Added queues %s to Edge Worker host %s by %s.", queues,
args.edge_hostname, getuser())
except TypeError as e:
- logger.error(str(e))
- raise SystemExit
+ raise SystemExit(str(e))
@cli_utils.action_cli(check_db=False)
@@ -428,8 +427,7 @@ def remove_worker_queues(args) -> None:
"Removed queues %s from Edge Worker host %s by %s.", queues,
args.edge_hostname, getuser()
)
except TypeError as e:
- logger.error(str(e))
- raise SystemExit
+ raise SystemExit(str(e))
@cli_utils.action_cli(check_db=False)
@@ -452,5 +450,4 @@ def set_remote_worker_concurrency(args) -> None:
getuser(),
)
except TypeError as e:
- logger.error(str(e))
- raise SystemExit
+ raise SystemExit(str(e))
diff --git a/providers/edge3/tests/unit/edge3/cli/test_edge_command.py
b/providers/edge3/tests/unit/edge3/cli/test_edge_command.py
index 0c0bad84fb9..5067f7584f2 100644
--- a/providers/edge3/tests/unit/edge3/cli/test_edge_command.py
+++ b/providers/edge3/tests/unit/edge3/cli/test_edge_command.py
@@ -14,5 +14,80 @@
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
+from __future__ import annotations
-# here is nothing to test, all is CLI wrapper only
+from argparse import Namespace
+from unittest import mock
+
+import pytest
+
+from airflow.providers.edge3.cli import edge_command
+
+# ``@cli_utils.action_cli`` writes a Log row via ``cli_action_loggers`` on
+# every invocation, which touches the metadata DB even with ``check_db=False``.
+# Same reason ``airflow-core/tests/unit/utils/test_cli_util.py`` marks its
+# module ``db_test``.
+pytestmark = pytest.mark.db_test
+
+
+class TestWorkerAdminExitCodes:
+ """CLI worker-admin subcommands must exit non-zero when the model
raises."""
+
+ @pytest.fixture(autouse=True)
+ def _stub_db_checks(self):
+ with (
+ mock.patch.object(edge_command, "_check_valid_db_connection"),
+ mock.patch.object(edge_command, "_check_if_registered_edge_host"),
+ ):
+ yield
+
+ @pytest.mark.parametrize(
+ ("cli_func", "model_attr", "args"),
+ [
+ (
+ edge_command.add_worker_queues,
+ "add_worker_queues",
+ Namespace(edge_hostname="worker-1", queues="q1,q2"),
+ ),
+ (
+ edge_command.remove_worker_queues,
+ "remove_worker_queues",
+ Namespace(edge_hostname="worker-1", queues="q1"),
+ ),
+ (
+ edge_command.set_remote_worker_concurrency,
+ "set_worker_concurrency",
+ Namespace(edge_hostname="worker-1", concurrency=8),
+ ),
+ (
+ edge_command.remote_worker_update_maintenance_comment,
+ "change_maintenance_comment",
+ Namespace(edge_hostname="worker-1", comments="short break"),
+ ),
+ (
+ edge_command.remove_remote_worker,
+ "remove_worker",
+ Namespace(edge_hostname="worker-1"),
+ ),
+ ],
+ ids=[
+ "add_worker_queues",
+ "remove_worker_queues",
+ "set_remote_worker_concurrency",
+ "remote_worker_update_maintenance_comment",
+ "remove_remote_worker",
+ ],
+ )
+ def test_exits_non_zero_when_model_raises_type_error(self, cli_func,
model_attr, args):
+ message = "Cannot mutate worker in OFFLINE state!"
+ with mock.patch(
+ f"airflow.providers.edge3.models.edge_worker.{model_attr}",
+ side_effect=TypeError(message),
+ ):
+ with pytest.raises(SystemExit) as exc_info:
+ cli_func(args)
+ # SystemExit carries the model's error message as its `code`, which the
+ # interpreter prints to stderr and translates to exit status 1. A bare
+ # `raise SystemExit` would leave code=None (exit 0), which is the bug
+ # this test guards against.
+ assert exc_info.value.code == message
