From 687acf130ab7df8b871716fa3d2753afbcb97516 Mon Sep 17 00:00:00 2001 From: Tal Rostoker Date: Wed, 28 Oct 2026 08:38:00 +0000 Subject: [PATCH 1/3] Add wait support for job start --- .../unreleased/added-20261026-191106.yaml | 6 + docs/commands/jobs/index.md | 11 +- docs/examples/job_examples.md | 8 + src/fabric_cli/commands/jobs/fab_jobs.py | 4 +- src/fabric_cli/errors/common.py | 4 + src/fabric_cli/parsers/fab_jobs_parser.py | 24 ++- src/fabric_cli/utils/fab_cmd_job_utils.py | 23 ++- tests/test_core/test_fab_jobs.py | 165 ++++++++++++++++++ tests/test_utils/test_fab_cmd_job_utils.py | 27 +++ 9 files changed, 265 insertions(+), 7 deletions(-) create mode 100644 .changes/unreleased/added-20261026-191106.yaml create mode 100644 tests/test_core/test_fab_jobs.py diff --git a/.changes/unreleased/added-20261026-191106.yaml b/.changes/unreleased/added-20261026-191106.yaml new file mode 100644 index 000000000..912ec4ad7 --- /dev/null +++ b/.changes/unreleased/added-20261026-191106.yaml @@ -0,0 +1,6 @@ +kind: added +body: Add --wait (-w), --timeout, and --polling_interval to job start to wait for completion like job run. +time: 2026-10-26T19:11:06.155500618Z +custom: + Author: tarostok_microsoft + AuthorLink: https://github.com/tarostok diff --git a/docs/commands/jobs/index.md b/docs/commands/jobs/index.md index 3ecd3c25d..50bda5f9c 100644 --- a/docs/commands/jobs/index.md +++ b/docs/commands/jobs/index.md @@ -13,7 +13,7 @@ The `job` commands provide tools for starting, running, monitoring, and scheduli | Command | Description | Usage | |-----------------|---------------------------|-----------------------------------------------------------------------| -| `job start` | Start an item (async) | `job start [-P ] [-C ] [-i ]` | +| `job start` | Start an item (async by default) | `job start [-P ] [-C ] [-i ] [--wait [--timeout ] [--polling_interval ]]` | | `job run` | Run an item (sync) | `job run [-P ] [-C ] [-i ] [--timeout ]` | | `job run-cancel`| Cancel an item run | `job run-cancel --id [--wait]` | | `job run-list` | List item or scheduled job runs | `job run-list [--schedule]` | @@ -27,11 +27,13 @@ The `job` commands provide tools for starting, running, monitoring, and scheduli ### start Start an item asynchronously and return immediately with a job ID. +Use `--wait` (or `-w`) to wait for completion with the same completion and +failure handling as `job run`. **Usage:** ``` -fab job start [-P ] [-C ] [-i ] +fab job start [-P ] [-C ] [-i ] [--wait [--timeout ] [--polling_interval ]] ``` **Parameters:** @@ -40,12 +42,17 @@ fab job start [-P ] [-C ] [-i ] - `-P, --params`: Parameters in name:type=value format, comma-separated. Optional. - `-C, --config`: JSON configuration for notebooks. Optional. - `-i, --input`: JSON payload for additional configuration. Optional. +- `-w, --wait`: Wait for the job to complete. Optional. +- `--timeout`: Maximum time to wait for completion in seconds. Requires `--wait`. Optional. +- `--polling_interval`: Custom job status polling interval in seconds. Requires `--wait`. Optional. **Examples:** ``` fab job start pipeline1.datapipeline fab job start notebook1.notebook -P param1:string=value1,param2:int=42 fab job start notebook1.notebook -C '{"spark.executor.memory": "4g"}' +fab job start ws1.Workspace/pipeline1.DataPipeline --wait +fab job start ws1.Workspace/notebook1.Notebook --wait --timeout 600 ``` --- diff --git a/docs/examples/job_examples.md b/docs/examples/job_examples.md index 8ac00f9c6..1187a3292 100644 --- a/docs/examples/job_examples.md +++ b/docs/examples/job_examples.md @@ -22,6 +22,14 @@ Execute a job synchronously, waiting for it to complete before returning control fab job run ws1.Workspace/pip1.DataPipeline ``` +Alternatively, use `job start --wait` (or `-w`) to wait for completion with the +same completion and failure handling as `job run`. `--timeout` and +`--polling_interval` are also supported with `--wait`. + +``` +fab job start ws1.Workspace/pip1.DataPipeline --wait --timeout 120 +``` + #### Run Job with Timeout Execute a job with a specific timeout limit in seconds. diff --git a/src/fabric_cli/commands/jobs/fab_jobs.py b/src/fabric_cli/commands/jobs/fab_jobs.py index d19db42f3..9b5ceb3c2 100644 --- a/src/fabric_cli/commands/jobs/fab_jobs.py +++ b/src/fabric_cli/commands/jobs/fab_jobs.py @@ -29,8 +29,8 @@ def run_command(args: Namespace) -> None: utils_job.add_item_props_to_args(args, context) utils_job.build_config_from_args(args, context) if args.jobs_command == "start": - fab_ui.print_grey(f"Starting job (async) for '{args.item}'...") - args.wait = False + mode = "sync" if args.wait else "async" + fab_ui.print_grey(f"Starting job ({mode}) for '{args.item}'...") if args.jobs_command == "run": fab_ui.print_grey(f"Running job (sync) for '{args.item}'...") args.wait = True diff --git a/src/fabric_cli/errors/common.py b/src/fabric_cli/errors/common.py index 6fcfc0009..cb646aa9d 100644 --- a/src/fabric_cli/errors/common.py +++ b/src/fabric_cli/errors/common.py @@ -162,6 +162,10 @@ def universal_security_disabled(name: str) -> str: def job_instance_failed(job_id: str) -> str: return f"Job instance '{job_id}' Failed" + @staticmethod + def options_require_wait(options: list[str]) -> str: + return f"{', '.join(options)} can only be used with --wait" + @staticmethod def file_not_accessible(file_path: str) -> str: return f"Cannot access '{file_path}'. No such file or directory" diff --git a/src/fabric_cli/parsers/fab_jobs_parser.py b/src/fabric_cli/parsers/fab_jobs_parser.py index f7343497a..84c03aeb5 100644 --- a/src/fabric_cli/parsers/fab_jobs_parser.py +++ b/src/fabric_cli/parsers/fab_jobs_parser.py @@ -13,7 +13,7 @@ commands = { "Commands": { - "start": "Start an item asynchronously.", + "start": "Start an item asynchronously, or wait for completion with --wait.", "run": "Run an item synchronously.", "run-cancel": "Cancel a running or scheduled job.", "run-list": "Retrieve the status of an item or scheduled job run.", @@ -35,7 +35,11 @@ def register_parser(subparsers: _SubParsersAction) -> None: "# run a pipeline async", "$ job start pip1.DataPipeline\n", "# run a notebook async", - "$ job start nb1.Notebook --input ", + "$ job start nb1.Notebook --input \n", + "# start a pipeline and wait for completion", + "$ job start pip1.DataPipeline --wait\n", + "# start a pipeline and wait with a 60 second timeout", + "$ job start pip1.DataPipeline --wait --timeout 60", ] start_parser = jobs_subparsers.add_parser( @@ -62,6 +66,22 @@ def register_parser(subparsers: _SubParsersAction) -> None: start_parser.add_argument( "-i", "--input", nargs="+", help="JSON payload, inline or path. Optional" ) + start_parser.add_argument( + "-w", "--wait", action="store_true", help="Wait for the job to complete. Optional" + ) + start_parser.add_argument( + "--timeout", + metavar="", + help="Timeout in seconds. Requires --wait. Optional", + type=int, + ) + start_parser.add_argument( + "--polling_interval", + metavar="", + dest="polling_interval", + help="Custom job status polling interval in seconds. Requires --wait. Optional", + type=validate_positive_int, + ) start_parser.usage = f"{utils_error_parser.get_usage_prog(start_parser)}" start_parser.set_defaults(func=lazy_command(_jobs_module_path, 'run_command')) diff --git a/src/fabric_cli/utils/fab_cmd_job_utils.py b/src/fabric_cli/utils/fab_cmd_job_utils.py index df4cc361b..e4c37eb4e 100644 --- a/src/fabric_cli/utils/fab_cmd_job_utils.py +++ b/src/fabric_cli/utils/fab_cmd_job_utils.py @@ -424,10 +424,31 @@ def _process_params(args: Namespace, item: Item) -> None: def validate_timeout_polling_interval(args: Namespace) -> None: - """Validate that polling interval is not greater than or equal to timeout.""" + """Validate timeout and polling interval options for job execution. + + Wait options require --wait for job start, and the polling + interval must be lower than the timeout. + """ timeout = getattr(args, "timeout", None) polling_interval = getattr(args, "polling_interval", None) + if getattr(args, "jobs_command", None) == "start" and not getattr( + args, "wait", False + ): + options = [ + option + for option, value in ( + ("--timeout", timeout), + ("--polling_interval", polling_interval), + ) + if value is not None + ] + if options: + raise FabricCLIError( + ErrorMessages.Common.options_require_wait(options), + fab_constant.ERROR_INVALID_INPUT, + ) + if ( timeout is not None and polling_interval is not None diff --git a/tests/test_core/test_fab_jobs.py b/tests/test_core/test_fab_jobs.py new file mode 100644 index 000000000..ba7666995 --- /dev/null +++ b/tests/test_core/test_fab_jobs.py @@ -0,0 +1,165 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Unit tests for job start waiting and job run compatibility.""" + +import inspect +import json +from argparse import Namespace +from unittest.mock import Mock + +import pytest + +from fabric_cli.commands.jobs import fab_jobs +from fabric_cli.core import fab_constant +from fabric_cli.core.fab_exceptions import FabricCLIError +from fabric_cli.core.fab_parser_setup import CustomArgumentParser +from fabric_cli.core.hiearchy.fab_item import Item +from fabric_cli.parsers import fab_jobs_parser + + +def _parse_job_args(command: str, flags: list[str]) -> Namespace: + parser = CustomArgumentParser() + fab_jobs_parser.register_parser(parser.add_subparsers(dest="command")) + args = parser.parse_args(["job", command, "ws.Workspace/nb.Notebook", *flags]) + args.item = "nb.Notebook" + args.output_format = "json" + args.command_path = f"job {command}" + return args + + +@pytest.fixture +def job_mocks(monkeypatch: pytest.MonkeyPatch) -> Namespace: + """Isolate context resolution, HTTP requests, output, and polling delays.""" + item = Mock(spec=Item) + monkeypatch.setattr( + fab_jobs.handle_context, "get_command_context", Mock(return_value=item) + ) + monkeypatch.setattr(fab_jobs.utils_job, "add_item_props_to_args", Mock()) + monkeypatch.setattr(fab_jobs.utils_job, "build_config_from_args", Mock()) + + response = Mock(status_code=202, headers={}) + start = Mock(return_value=(response, "test-job-id")) + poll = Mock( + side_effect=[ + Mock(status_code=200, headers={}, text=json.dumps({"status": status})) + for status in ("InProgress", "Completed") + ] + ) + sleep = Mock() + output = Mock() + monkeypatch.setattr(fab_jobs.jobs_run.jobs_api, "run_on_demand_item_job", start) + monkeypatch.setattr(fab_jobs.utils_job.jobs_api, "get_item_job_instance", poll) + monkeypatch.setattr(fab_jobs.utils_job.time, "sleep", sleep) + monkeypatch.setattr(fab_jobs.fab_ui, "print_grey", Mock()) + monkeypatch.setattr(fab_jobs.fab_ui, "print_progress", Mock()) + monkeypatch.setattr(fab_jobs.fab_ui, "print_output_format", output) + monkeypatch.setattr(fab_jobs.fab_ui, "print_entries_unix_style", Mock()) + return Namespace(start=start, poll=poll, sleep=sleep, output=output) + + +@pytest.mark.parametrize( + ("command", "flags", "wait"), + [ + ("start", [], False), + ("start", ["--wait"], True), + ("start", ["-w"], True), + ("run", [], True), + ], +) +def test_job_wait_behavior( + command: str, flags: list[str], wait: bool, job_mocks: Namespace +) -> None: + args = _parse_job_args(command, flags) + + inspect.unwrap(fab_jobs.run_command)(args) + + assert args.wait is wait + job_mocks.start.assert_called_once_with(args, None) + assert job_mocks.poll.call_count == (2 if wait else 0) + assert job_mocks.sleep.call_count == (2 if wait else 0) + job_mocks.output.assert_called_once() + assert job_mocks.output.call_args.kwargs["data"] == {"id": "test-job-id"} + assert job_mocks.output.call_args.kwargs["message"] == ( + f"Job instance 'test-job-id' {'completed' if wait else 'created'}" + ) + + +@pytest.mark.parametrize("command", ["start", "run"]) +def test_job_wait_reports_failure(command: str, job_mocks: Namespace) -> None: + args = _parse_job_args(command, ["--wait"] if command == "start" else []) + job_mocks.poll.side_effect = None + job_mocks.poll.return_value = Mock( + status_code=200, headers={}, text=json.dumps({"status": "Failed"}) + ) + + with pytest.raises(FabricCLIError) as exc_info: + inspect.unwrap(fab_jobs.run_command)(args) + + assert exc_info.value.status_code == fab_constant.ERROR_JOB_FAILED + job_mocks.output.assert_not_called() + + +@pytest.mark.parametrize( + ("flags", "expected"), + [ + (["--timeout", "60"], "--timeout can only be used with --wait"), + ( + ["--polling_interval", "10"], + "--polling_interval can only be used with --wait", + ), + ( + ["--timeout", "60", "--polling_interval", "10"], + "--timeout, --polling_interval can only be used with --wait", + ), + ], +) +def test_job_start_wait_options_without_wait_failure( + flags: list[str], expected: str, job_mocks: Namespace +) -> None: + args = _parse_job_args("start", flags) + + with pytest.raises(FabricCLIError) as exc_info: + inspect.unwrap(fab_jobs.run_command)(args) + + assert exc_info.value.status_code == fab_constant.ERROR_INVALID_INPUT + assert exc_info.value.message == expected + job_mocks.start.assert_not_called() + + +def test_job_start_wait_timeout_success( + job_mocks: Namespace, monkeypatch: pytest.MonkeyPatch +) -> None: + args = _parse_job_args( + "start", ["--wait", "--timeout", "60", "--polling_interval", "10"] + ) + wait = Mock() + monkeypatch.setattr(fab_jobs.jobs_run.utils_job, "wait_for_job_completion", wait) + + inspect.unwrap(fab_jobs.run_command)(args) + + wait.assert_called_once() + assert wait.call_args.kwargs["timeout"] == 60 + assert wait.call_args.kwargs["custom_polling_interval"] == 10 + + +def test_job_start_wait_timeout_cancels_job( + job_mocks: Namespace, + monkeypatch: pytest.MonkeyPatch, + mock_fab_set_state_config, +) -> None: + args = _parse_job_args("start", ["--wait", "--timeout", "5"]) + mock_fab_set_state_config(fab_constant.FAB_JOB_CANCEL_ONTIMEOUT, "true") + monkeypatch.setattr( + fab_jobs.jobs_run.utils_job, + "wait_for_job_completion", + Mock(side_effect=TimeoutError("timed out")), + ) + monkeypatch.setattr(fab_jobs.fab_ui, "print_warning", Mock()) + cancel = Mock(return_value=Mock(status_code=202)) + monkeypatch.setattr(fab_jobs.jobs_run.jobs_api, "cancel_item_job_instance", cancel) + + inspect.unwrap(fab_jobs.run_command)(args) + + cancel.assert_called_once_with(args) + assert args.instance_id == "test-job-id" diff --git a/tests/test_utils/test_fab_cmd_job_utils.py b/tests/test_utils/test_fab_cmd_job_utils.py index 9ea42af90..c99d7eab1 100644 --- a/tests/test_utils/test_fab_cmd_job_utils.py +++ b/tests/test_utils/test_fab_cmd_job_utils.py @@ -267,5 +267,32 @@ def test_validate_timeout_polling_interval_none_values_success(): validate_timeout_polling_interval(args) +@pytest.mark.parametrize("wait_args", [{}, {"wait": False}]) +def test_validate_timeout_polling_interval_without_wait_failure( + wait_args: dict[str, bool], +) -> None: + args = Namespace( + jobs_command="start", timeout=60, polling_interval=None, **wait_args + ) + + with pytest.raises(FabricCLIError) as exc_info: + validate_timeout_polling_interval(args) + + assert exc_info.value.status_code == fab_constant.ERROR_INVALID_INPUT + assert exc_info.value.message == "--timeout can only be used with --wait" + + +def test_validate_timeout_polling_interval_with_wait_success(): + args = Namespace(jobs_command="start", wait=True, timeout=60, polling_interval=10) + + validate_timeout_polling_interval(args) + + +def test_validate_timeout_polling_interval_run_with_wait_false_success() -> None: + args = Namespace(jobs_command="run", wait=False, timeout=60, polling_interval=10) + + validate_timeout_polling_interval(args) + + if __name__ == "__main__": pytest.main([__file__]) From b54e49e015dadb83f01842dbd5435a5c9787bb13 Mon Sep 17 00:00:00 2001 From: Tal Rostoker Date: Sun, 1 Nov 2026 10:04:04 +0000 Subject: [PATCH 2/3] fix PR comments. --- docs/commands/jobs/index.md | 4 ++-- src/fabric_cli/errors/common.py | 4 ++++ src/fabric_cli/parsers/fab_jobs_parser.py | 3 +-- src/fabric_cli/utils/fab_cmd_job_utils.py | 8 +++++++- 4 files changed, 14 insertions(+), 5 deletions(-) diff --git a/docs/commands/jobs/index.md b/docs/commands/jobs/index.md index 50bda5f9c..2961cd1af 100644 --- a/docs/commands/jobs/index.md +++ b/docs/commands/jobs/index.md @@ -43,7 +43,7 @@ fab job start [-P ] [-C ] [-i ] [--w - `-C, --config`: JSON configuration for notebooks. Optional. - `-i, --input`: JSON payload for additional configuration. Optional. - `-w, --wait`: Wait for the job to complete. Optional. -- `--timeout`: Maximum time to wait for completion in seconds. Requires `--wait`. Optional. +- `--timeout`: Maximum time to wait for completion in seconds (0 or greater). Requires `--wait`. Optional. - `--polling_interval`: Custom job status polling interval in seconds. Requires `--wait`. Optional. **Examples:** @@ -73,7 +73,7 @@ fab job run [-P ] [-C ] [-i ] [--timeout str: def options_require_wait(options: list[str]) -> str: return f"{', '.join(options)} can only be used with --wait" + @staticmethod + def negative_job_timeout(timeout: int) -> str: + return f"Timeout ({timeout}s) must be greater than or equal to 0" + @staticmethod def file_not_accessible(file_path: str) -> str: return f"Cannot access '{file_path}'. No such file or directory" diff --git a/src/fabric_cli/parsers/fab_jobs_parser.py b/src/fabric_cli/parsers/fab_jobs_parser.py index 84c03aeb5..150f54079 100644 --- a/src/fabric_cli/parsers/fab_jobs_parser.py +++ b/src/fabric_cli/parsers/fab_jobs_parser.py @@ -72,13 +72,12 @@ def register_parser(subparsers: _SubParsersAction) -> None: start_parser.add_argument( "--timeout", metavar="", - help="Timeout in seconds. Requires --wait. Optional", + help="Timeout in seconds (0 or greater). Requires --wait. Optional", type=int, ) start_parser.add_argument( "--polling_interval", metavar="", - dest="polling_interval", help="Custom job status polling interval in seconds. Requires --wait. Optional", type=validate_positive_int, ) diff --git a/src/fabric_cli/utils/fab_cmd_job_utils.py b/src/fabric_cli/utils/fab_cmd_job_utils.py index e4c37eb4e..ba103747f 100644 --- a/src/fabric_cli/utils/fab_cmd_job_utils.py +++ b/src/fabric_cli/utils/fab_cmd_job_utils.py @@ -427,7 +427,7 @@ def validate_timeout_polling_interval(args: Namespace) -> None: """Validate timeout and polling interval options for job execution. Wait options require --wait for job start, and the polling - interval must be lower than the timeout. + interval must be lower than the timeout. Timeouts must be non-negative. """ timeout = getattr(args, "timeout", None) polling_interval = getattr(args, "polling_interval", None) @@ -449,6 +449,12 @@ def validate_timeout_polling_interval(args: Namespace) -> None: fab_constant.ERROR_INVALID_INPUT, ) + if timeout is not None and timeout < 0: + raise FabricCLIError( + ErrorMessages.Common.negative_job_timeout(timeout), + fab_constant.ERROR_INVALID_INPUT, + ) + if ( timeout is not None and polling_interval is not None From 544c5bcb2d4648dc6810d3260448b9cc9417cdea Mon Sep 17 00:00:00 2001 From: Tal Rostoker Date: Sun, 1 Nov 2026 11:30:03 +0000 Subject: [PATCH 3/3] Additional tests --- tests/test_core/test_fab_jobs.py | 36 ++++++++++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/tests/test_core/test_fab_jobs.py b/tests/test_core/test_fab_jobs.py index ba7666995..10ce09075 100644 --- a/tests/test_core/test_fab_jobs.py +++ b/tests/test_core/test_fab_jobs.py @@ -127,6 +127,42 @@ def test_job_start_wait_options_without_wait_failure( job_mocks.start.assert_not_called() +@pytest.mark.parametrize("command", ["start", "run"]) +def test_job_negative_timeout_failure(command: str, job_mocks: Namespace) -> None: + flags = ["--wait"] if command == "start" else [] + args = _parse_job_args(command, [*flags, "--timeout", "-1"]) + + with pytest.raises(FabricCLIError) as exc_info: + inspect.unwrap(fab_jobs.run_command)(args) + + assert exc_info.value.status_code == fab_constant.ERROR_INVALID_INPUT + job_mocks.start.assert_not_called() + job_mocks.poll.assert_not_called() + job_mocks.sleep.assert_not_called() + + +@pytest.mark.parametrize("command", ["start", "run"]) +def test_job_zero_timeout_success( + command: str, job_mocks: Namespace, monkeypatch: pytest.MonkeyPatch +) -> None: + flags = ["--wait"] if command == "start" else [] + args = _parse_job_args(command, [*flags, "--timeout", "0"]) + monkeypatch.setattr( + fab_jobs.jobs_run.config, "get_config", Mock(return_value="false") + ) + warning = Mock() + monkeypatch.setattr(fab_jobs.fab_ui, "print_warning", warning) + + inspect.unwrap(fab_jobs.run_command)(args) + + job_mocks.start.assert_called_once_with(args, None) + job_mocks.sleep.assert_called_once_with(0) + job_mocks.poll.assert_not_called() + warning.assert_called_once_with( + "Job instance 'test-job-id' timed out after 0 seconds" + ) + + def test_job_start_wait_timeout_success( job_mocks: Namespace, monkeypatch: pytest.MonkeyPatch ) -> None: