From 2e23b799feca3dead203cb3c8fc0b293864be46f Mon Sep 17 00:00:00 2001 From: Leo Ueno Date: Mon, 24 Aug 2026 01:43:57 -0700 Subject: [PATCH] Add Batch Processing CLI commands --- CLI-COMMANDS.md | 21 ++ roboflow/adapters/rfapi.py | 119 +++++++ roboflow/cli/__init__.py | 2 +- roboflow/cli/handlers/batch.py | 300 ++++++++++++++++-- tests/adapters/test_rfapi_batch_processing.py | 56 ++++ tests/cli/test_batch_handler.py | 187 ++++++++++- tests/cli/test_completion_handler.py | 1 - 7 files changed, 640 insertions(+), 46 deletions(-) create mode 100644 tests/adapters/test_rfapi_batch_processing.py diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 4593b802..03acb27e 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -47,6 +47,27 @@ roboflow download my-workspace/my-project/3 -f coco # alias roboflow infer photo.jpg -m my-project/3 ``` +### Batch Process Asset Library images + +```bash +# Exact reviewed selection (CPU is the product default): +roboflow batch create --workflow inspect-defects --image-ids img_1,img_2 + +# Or every current match for a structured RoboQL filter: +roboflow batch create --workflow inspect-defects --query "tag:night-shift" + +# Monitor and control the durable job: +roboflow batch status +roboflow batch list +roboflow batch abort +roboflow batch restart +``` + +The create response includes `taskId`, `jobId`, and `requestId`. A job continues if the terminal +closes. If a create request has an ambiguous network result, retry with the same `--request-id` to +avoid a duplicate. Local folders must first be uploaded into Roboflow; Batch Processing never sends +local file contents through an Agent or CLI job-configuration request. + ### Train, monitor, cancel, stop ```bash diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index e2631122..236725d5 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -1376,6 +1376,125 @@ def get_video_job_status(api_key, job_id): return response.json() +# --------------------------------------------------------------------------- +# Batch Processing (Asset Library orchestration) +# --------------------------------------------------------------------------- + + +def _batch_processing_url(workspace_url, suffix=""): + return f"{API_URL}/batch-processing/v1/external/{workspace_url}/asset-library/jobs{suffix}" + + +def _batch_processing_headers(api_key): + # Keep credentials out of URLs, proxy logs, and shell history. validateToken supports Bearer. + return {"Authorization": f"Bearer {api_key}"} + + +def _raise_for_batch_processing_response(response): + message = response.text + try: + body = response.json() + if isinstance(body, dict): + error = body.get("error") + if isinstance(error, dict): + message = error.get("message") or error.get("hint") or message + elif error: + message = str(error) + else: + message = body.get("message") or message + except (TypeError, ValueError): + pass + raise RoboflowError(message, status_code=response.status_code) + + +def create_asset_library_batch_job( + api_key, + workspace_url, + *, + workflow_id, + idempotency_key, + image_ids=None, + query=None, + machine_type="cpu", + display_name=None, +): + """Queue a published Workflow over an exact Asset Library selection.""" + payload = { + "workflowId": workflow_id, + "idempotencyKey": idempotency_key, + "machineType": machine_type, + } + if image_ids is not None: + payload["imageIds"] = image_ids + if query is not None: + payload["query"] = query + if display_name: + payload["displayName"] = display_name + response = requests.post( + _batch_processing_url(workspace_url), + headers=_batch_processing_headers(api_key), + json=payload, + ) + if response.status_code != 202: + _raise_for_batch_processing_response(response) + return response.json() + + +def list_batch_processing_jobs(api_key, workspace_url, *, page_size=10, next_page_token=None, search=None): + """List durable Batch Processing jobs in a workspace.""" + params = {"pageSize": page_size} + if next_page_token: + params["nextPageToken"] = next_page_token + if search: + params["search"] = search + response = requests.get( + _batch_processing_url(workspace_url), + headers=_batch_processing_headers(api_key), + params=params, + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + +def get_batch_processing_job(api_key, workspace_url, job_id): + """Get current metadata for one Batch Processing job.""" + encoded = quote(job_id, safe="") + response = requests.get( + _batch_processing_url(workspace_url, f"/{encoded}"), + headers=_batch_processing_headers(api_key), + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + +def abort_batch_processing_job(api_key, workspace_url, job_id): + """Abort one Batch Processing job.""" + encoded = quote(job_id, safe="") + response = requests.post( + _batch_processing_url(workspace_url, f"/{encoded}/abort"), + headers=_batch_processing_headers(api_key), + json={}, + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + +def restart_batch_processing_job(api_key, workspace_url, job_id): + """Restart one Batch Processing job with its existing configuration.""" + encoded = quote(job_id, safe="") + response = requests.post( + _batch_processing_url(workspace_url, f"/{encoded}/restart"), + headers=_batch_processing_headers(api_key), + json={}, + ) + if response.status_code != 200: + _raise_for_batch_processing_response(response) + return response.json() + + # --------------------------------------------------------------------------- # Phase 2: Universe search # --------------------------------------------------------------------------- diff --git a/roboflow/cli/__init__.py b/roboflow/cli/__init__.py index 54754a08..3befa965 100644 --- a/roboflow/cli/__init__.py +++ b/roboflow/cli/__init__.py @@ -213,7 +213,7 @@ def _walk(group: Any, prefix: str = "") -> None: app.add_typer(api_key_app, name="api-key") app.add_typer(asynctasks_app, name="asynctasks") app.add_typer(auth_app, name="auth") -app.add_typer(batch_app, name="batch", hidden=True) # All stubs — hidden until implemented +app.add_typer(batch_app, name="batch") app.add_typer(completion_app, name="completion") app.add_typer(deployment_app, name="deployment") app.add_typer(device_app, name="device") diff --git a/roboflow/cli/handlers/batch.py b/roboflow/cli/handlers/batch.py index 31d24647..dfd29940 100644 --- a/roboflow/cli/handlers/batch.py +++ b/roboflow/cli/handlers/batch.py @@ -1,33 +1,64 @@ -"""Batch processing commands.""" +"""Batch Processing commands backed by Roboflow's durable workspace jobs.""" from __future__ import annotations +import re +import uuid +from enum import Enum from typing import Annotated, Optional import typer from roboflow.cli._compat import SortedGroup, ctx_to_args -batch_app = typer.Typer(cls=SortedGroup, help="Batch processing operations", no_args_is_help=True) +batch_app = typer.Typer(cls=SortedGroup, help="Run and manage Batch Processing jobs", no_args_is_help=True) -def _stub(args) -> None: # noqa: ANN001 - from roboflow.cli._output import output_error +class BatchMachine(str, Enum): + """Execution pools exposed by the Asset Library Batch Processing surface.""" - output_error(args, "This command is not yet implemented.", hint="Coming soon.", exit_code=1) + CPU = "cpu" + GPU = "gpu" @batch_app.command("create") def create( ctx: typer.Context, - workflow: Annotated[str, typer.Option(help="Workflow ID to run")], - input: Annotated[str, typer.Option(help="Input path (image directory or video file)")], - model: Annotated[Optional[str], typer.Option(help="Model ID override (default: workflow model)")] = None, - output_dir: Annotated[Optional[str], typer.Option("--output", help="Output directory for results")] = None, + workflow: Annotated[str, typer.Option(help="Published Workflow ID to run")], + image_ids: Annotated[ + Optional[str], + typer.Option("--image-ids", help="Comma-separated exact Asset Library image IDs"), + ] = None, + query: Annotated[ + Optional[str], + typer.Option(help="Reviewed structured RoboQL filter selecting all current matches"), + ] = None, + all_images: Annotated[ + bool, + typer.Option("--all", help="Explicitly run on the entire Asset Library"), + ] = False, + machine: Annotated[ + BatchMachine, + typer.Option(help="Execution machine; defaults to the product UI default"), + ] = BatchMachine.CPU, + name: Annotated[Optional[str], typer.Option(help="Optional user-facing job name")] = None, + request_id: Annotated[ + Optional[str], + typer.Option(help="Stable idempotency key; reuse only when retrying this same launch"), + ] = None, ) -> None: - """Create a batch processing job.""" - args = ctx_to_args(ctx, workflow=workflow, input=input, model=model, output=output_dir) - _stub(args) + """Queue a Workflow over one exact or reviewed Asset Library selection.""" + args = ctx_to_args( + ctx, + workflow=workflow, + image_ids=image_ids, + query=query, + all_images=all_images, + machine=machine.value, + name=name, + request_id=request_id, + ) + _create(args) @batch_app.command("status") @@ -35,29 +66,244 @@ def status( ctx: typer.Context, job_id: Annotated[str, typer.Argument(help="Batch job ID")], ) -> None: - """Check batch job status.""" - args = ctx_to_args(ctx, job_id=job_id) - _stub(args) + """Show current durable job status and configuration.""" + _status(ctx_to_args(ctx, job_id=job_id)) @batch_app.command("list") def list_jobs( ctx: typer.Context, - status_filter: Annotated[ - Optional[str], typer.Option("--status", help="Filter by status (pending, running, completed, failed)") - ] = None, + page_size: Annotated[int, typer.Option(min=1, max=100, help="Jobs per page")] = 10, + next_page_token: Annotated[Optional[str], typer.Option(help="Pagination token")] = None, + search: Annotated[Optional[str], typer.Option(help="Search names, Workflows, and status text")] = None, ) -> None: - """List batch jobs.""" - args = ctx_to_args(ctx, status=status_filter) - _stub(args) + """List Batch Processing jobs.""" + _list(ctx_to_args(ctx, page_size=page_size, next_page_token=next_page_token, search=search)) -@batch_app.command("results") -def results( +@batch_app.command("abort") +def abort( ctx: typer.Context, job_id: Annotated[str, typer.Argument(help="Batch job ID")], - format: Annotated[Optional[str], typer.Option(help="Output format (json, csv)")] = None, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Confirm without prompting")] = False, ) -> None: - """Get batch job results.""" - args = ctx_to_args(ctx, job_id=job_id, format=format) - _stub(args) + """Abort a Batch Processing job.""" + _abort(ctx_to_args(ctx, job_id=job_id, yes=yes)) + + +@batch_app.command("restart") +def restart( + ctx: typer.Context, + job_id: Annotated[str, typer.Argument(help="Batch job ID")], + yes: Annotated[bool, typer.Option("--yes", "-y", help="Confirm credit-spending restart")] = False, +) -> None: + """Restart a Batch Processing job with its existing configuration.""" + _restart(ctx_to_args(ctx, job_id=job_id, yes=yes)) + + +def _resolve_ws_and_key(args): # noqa: ANN001 + from roboflow.cli._resolver import resolve_ws_and_key + + return resolve_ws_and_key(args) + + +def _parse_image_ids(raw: Optional[str]) -> list[str]: + if raw is None: + return [] + return list(dict.fromkeys(part.strip() for part in raw.split(",") if part.strip())) + + +def _validate_job_id(args, job_id: str, *, label: str = "job ID") -> None: # noqa: ANN001 + from roboflow.cli._output import output_error + + if not re.fullmatch(r"[a-z0-9-]{1,20}", job_id): + output_error( + args, + f"{label} must be 1-20 lowercase letters, numbers, or hyphens.", + ) + + +def _create(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error, output_error + + ids = _parse_image_ids(args.image_ids) + selection_modes = int(bool(ids)) + int(args.query is not None) + int(args.all_images) + if selection_modes != 1: + output_error( + args, + "Choose exactly one selection: --image-ids, --query, or --all.", + hint="The CLI never broadens an omitted or ambiguous selection to the whole Asset Library.", + ) + return + if args.query is not None and not args.query.strip(): + output_error( + args, + "--query cannot be empty.", + hint="Use --all to explicitly select the entire Asset Library.", + ) + return + if len(ids) > 2048: + output_error(args, "At most 2048 explicit image IDs can be queued in one request.") + return + + request_id = args.request_id or str(uuid.uuid4()) + if not 8 <= len(request_id) <= 128 or not re.fullmatch(r"[A-Za-z0-9_-]+", request_id): + output_error( + args, + "--request-id must be 8-128 letters, numbers, underscores, or hyphens.", + ) + return + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + query = "" if args.all_images else args.query + + try: + result = rfapi.create_asset_library_batch_job( + api_key, + workspace, + workflow_id=args.workflow, + idempotency_key=request_id, + image_ids=ids or None, + query=query, + machine_type=args.machine, + display_name=args.name, + ) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + hint=f"Retry the same launch with --request-id {request_id}; use a new ID only for a new job intent.", + auth_hint="Check the API key has 'batch-processing:trigger' scope and access to the selected resources.", + ) + return + + result = {**result, "requestId": request_id} + text = ( + f"Queued {result.get('displayName') or result.get('jobId')}\n" + f"jobId={result.get('jobId')}\n" + f"taskId={result.get('taskId')}\n" + f"requestId={request_id}\n" + f"Next: roboflow asynctasks wait {result.get('taskId')}" + ) + output(args, result, text=text) + + +def _status(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error + + _validate_job_id(args, args.job_id) + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + try: + result = rfapi.get_batch_processing_job(api_key, workspace, args.job_id) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + auth_hint="Check the API key has 'batch-processing:read' scope.", + not_found_hint="Check the job ID and workspace.", + ) + return + + job = result.get("job", {}) + state = job.get("currentStage") or ("terminal" if job.get("isTerminal") else "queued") + output( + args, + result, + text=( + f"jobId={job.get('jobId', args.job_id)} state={state} " + f"terminal={job.get('isTerminal')} error={job.get('error')}" + ), + ) + + +def _list(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error, output_error + from roboflow.cli._table import format_table + + if args.next_page_token: + _validate_job_id(args, args.next_page_token, label="--next-page-token") + if args.search is not None and len(args.search) > 160: + output_error(args, "--search must be at most 160 characters.") + return + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + try: + result = rfapi.list_batch_processing_jobs( + api_key, + workspace, + page_size=args.page_size, + next_page_token=args.next_page_token, + search=args.search, + ) + except rfapi.RoboflowError as exc: + output_api_error(args, exc, auth_hint="Check the API key has 'batch-processing:read' scope.") + return + + rows = [ + { + "jobId": job.get("jobId", ""), + "name": job.get("name", ""), + "stage": job.get("currentStage") or ("terminal" if job.get("isTerminal") else "queued"), + "error": job.get("error", False), + "updated": job.get("lastUpdate", ""), + } + for job in result.get("jobs", []) + ] + table = format_table(rows, columns=["jobId", "name", "stage", "error", "updated"]) + if result.get("nextPageToken"): + table += f"\nNext page: --next-page-token {result['nextPageToken']}" + output(args, result, text=table) + + +def _abort(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import confirm_destructive + + _validate_job_id(args, args.job_id) + if not confirm_destructive(args, f"Abort Batch Processing job '{args.job_id}'?"): + return + _run_control_action(args, rfapi.abort_batch_processing_job, "Aborted") + + +def _restart(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import confirm_destructive + + _validate_job_id(args, args.job_id) + if not confirm_destructive( + args, + f"Restart Batch Processing job '{args.job_id}'? This can consume credits.", + ): + return + _run_control_action(args, rfapi.restart_batch_processing_job, "Restarted") + + +def _run_control_action(args, action, verb: str) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import output, output_api_error + + resolved = _resolve_ws_and_key(args) + if not resolved: + return + workspace, api_key = resolved + try: + result = action(api_key, workspace, args.job_id) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + auth_hint="Check the API key has 'batch-processing:trigger' scope.", + not_found_hint="Check the job ID and workspace.", + ) + return + output(args, result, text=f"{verb} Batch Processing job {args.job_id}.") diff --git a/tests/adapters/test_rfapi_batch_processing.py b/tests/adapters/test_rfapi_batch_processing.py new file mode 100644 index 00000000..ce65a2e4 --- /dev/null +++ b/tests/adapters/test_rfapi_batch_processing.py @@ -0,0 +1,56 @@ +"""HTTP contract tests for Asset Library Batch Processing adapters.""" + +from __future__ import annotations + +import unittest +from unittest.mock import Mock, patch + +from roboflow.adapters import rfapi + + +class TestBatchProcessingAdapter(unittest.TestCase): + @patch("roboflow.adapters.rfapi.requests.post") + def test_create_uses_bearer_auth_and_idempotent_payload(self, mock_post) -> None: + response = Mock(status_code=202) + response.json.return_value = {"status": "queued", "jobId": "al-123"} + mock_post.return_value = response + + result = rfapi.create_asset_library_batch_job( + "private-key", + "workspace-1", + workflow_id="workflow-1", + idempotency_key="request-123", + image_ids=["image-1"], + ) + + self.assertEqual(result["jobId"], "al-123") + _, kwargs = mock_post.call_args + self.assertEqual(kwargs["headers"], {"Authorization": "Bearer private-key"}) + self.assertNotIn("private-key", mock_post.call_args.args[0]) + self.assertEqual(kwargs["json"]["idempotencyKey"], "request-123") + + @patch("roboflow.adapters.rfapi.requests.get") + def test_status_encodes_untrusted_job_id(self, mock_get) -> None: + response = Mock(status_code=200) + response.json.return_value = {"status": "ok", "job": {"jobId": "bad/id"}} + mock_get.return_value = response + + rfapi.get_batch_processing_job("private-key", "workspace-1", "bad/id") + + self.assertTrue(mock_get.call_args.args[0].endswith("/bad%2Fid")) + + @patch("roboflow.adapters.rfapi.requests.get") + def test_error_preserves_http_status_for_cli_exit_codes(self, mock_get) -> None: + response = Mock(status_code=404, text='{"error":{"message":"Job not found"}}') + response.json.return_value = {"error": {"message": "Job not found"}} + mock_get.return_value = response + + with self.assertRaises(rfapi.RoboflowError) as ctx: + rfapi.get_batch_processing_job("private-key", "workspace-1", "missing") + + self.assertEqual(ctx.exception.status_code, 404) + self.assertEqual(str(ctx.exception), "Job not found") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/cli/test_batch_handler.py b/tests/cli/test_batch_handler.py index bfe773d1..bc605489 100644 --- a/tests/cli/test_batch_handler.py +++ b/tests/cli/test_batch_handler.py @@ -1,37 +1,190 @@ -"""Tests for the batch CLI handler.""" +"""Tests for the durable Batch Processing CLI.""" +from __future__ import annotations + +import json import unittest +from unittest.mock import patch from typer.testing import CliRunner from roboflow.cli import app runner = CliRunner() +BASE = ["--workspace", "workspace-1", "--api-key", "private-key"] class TestBatchRegistration(unittest.TestCase): - """Verify batch handler registers expected subcommands.""" + """Batch commands are public and documented by Typer.""" + + def test_batch_is_visible_in_root_help(self) -> None: + result = runner.invoke(app, ["--help"]) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("batch", result.output) + self.assertIn("Run and manage Batch Processing jobs", result.output) + + def test_batch_subcommands(self) -> None: + for verb in ("create", "status", "list", "abort", "restart"): + with self.subTest(verb=verb): + result = runner.invoke(app, ["batch", verb, "--help"]) + self.assertEqual(result.exit_code, 0, result.output) + + +class TestBatchCreate(unittest.TestCase): + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_create_exact_selection_has_stable_machine_output(self, mock_create) -> None: + mock_create.return_value = { + "status": "queued", + "taskId": "task-1", + "jobId": "al-123", + "batchId": "asset-library-123", + "displayName": "Night defects", + } + + result = runner.invoke( + app, + [ + *BASE, + "--json", + "batch", + "create", + "--workflow", + "workflow-1", + "--image-ids", + "image-1,image-1,image-2", + "--request-id", + "request-123", + ], + ) + + self.assertEqual(result.exit_code, 0, result.output) + payload = json.loads(result.output) + self.assertEqual(payload["requestId"], "request-123") + mock_create.assert_called_once_with( + "private-key", + "workspace-1", + workflow_id="workflow-1", + idempotency_key="request-123", + image_ids=["image-1", "image-2"], + query=None, + machine_type="cpu", + display_name=None, + ) + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_all_is_explicit_empty_query_not_an_omitted_selection(self, mock_create) -> None: + mock_create.return_value = {"taskId": "task-1", "jobId": "al-123"} + + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1", "--all"], + ) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(mock_create.call_args.kwargs["query"], "") + self.assertIsNone(mock_create.call_args.kwargs["image_ids"]) + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_ambiguous_selection_fails_closed(self, mock_create) -> None: + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1"], + ) + + self.assertEqual(result.exit_code, 1) + self.assertIn("exactly one selection", result.output) + mock_create.assert_not_called() + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_empty_query_does_not_implicitly_select_every_image(self, mock_create) -> None: + result = runner.invoke( + app, + [*BASE, "batch", "create", "--workflow", "workflow-1", "--query", ""], + ) + + self.assertEqual(result.exit_code, 1) + self.assertIn("--query cannot be empty", result.output) + self.assertIn("Use --all", result.output) + mock_create.assert_not_called() + + @patch("roboflow.adapters.rfapi.create_asset_library_batch_job") + def test_unsafe_request_id_fails_before_network(self, mock_create) -> None: + result = runner.invoke( + app, + [ + *BASE, + "batch", + "create", + "--workflow", + "workflow-1", + "--all", + "--request-id", + "unsafe/key", + ], + ) + + self.assertEqual(result.exit_code, 1) + self.assertIn("--request-id must be", result.output) + mock_create.assert_not_called() + + +class TestBatchLifecycle(unittest.TestCase): + @patch("roboflow.adapters.rfapi.get_batch_processing_job") + def test_status_rejects_malformed_job_id_before_network(self, mock_get) -> None: + result = runner.invoke(app, [*BASE, "batch", "status", "unsafe/job"]) + + self.assertEqual(result.exit_code, 1) + self.assertIn("job ID must be", result.output) + mock_get.assert_not_called() + + @patch("roboflow.adapters.rfapi.get_batch_processing_job") + def test_status_json_is_api_faithful(self, mock_get) -> None: + api_result = { + "status": "ok", + "job": {"jobId": "al-123", "currentStage": "inference", "isTerminal": False, "error": False}, + } + mock_get.return_value = api_result + + result = runner.invoke(app, [*BASE, "--json", "batch", "status", "al-123"]) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(json.loads(result.output), api_result) + + @patch("roboflow.adapters.rfapi.list_batch_processing_jobs") + def test_list_passes_pagination_and_search(self, mock_list) -> None: + mock_list.return_value = {"status": "ok", "jobs": [], "nextPageToken": None} + + result = runner.invoke( + app, + [*BASE, "batch", "list", "--page-size", "25", "--next-page-token", "next-1", "--search", "night"], + ) + + self.assertEqual(result.exit_code, 0, result.output) + mock_list.assert_called_once_with( + "private-key", + "workspace-1", + page_size=25, + next_page_token="next-1", + search="night", + ) - def test_batch_app_exists(self) -> None: - from roboflow.cli.handlers.batch import batch_app + @patch("roboflow.adapters.rfapi.abort_batch_processing_job") + def test_abort_requires_and_honors_explicit_confirmation(self, mock_abort) -> None: + mock_abort.return_value = {"status": "ok", "jobId": "al-123"} - self.assertIsNotNone(batch_app) + result = runner.invoke(app, [*BASE, "batch", "abort", "al-123", "--yes"]) - def test_batch_create_exists(self) -> None: - result = runner.invoke(app, ["batch", "create", "--help"]) - self.assertEqual(result.exit_code, 0) + self.assertEqual(result.exit_code, 0, result.output) + mock_abort.assert_called_once_with("private-key", "workspace-1", "al-123") - def test_batch_status_exists(self) -> None: - result = runner.invoke(app, ["batch", "status", "--help"]) - self.assertEqual(result.exit_code, 0) + @patch("roboflow.adapters.rfapi.restart_batch_processing_job") + def test_restart_requires_and_honors_credit_confirmation(self, mock_restart) -> None: + mock_restart.return_value = {"status": "ok", "jobId": "al-123"} - def test_batch_list_exists(self) -> None: - result = runner.invoke(app, ["batch", "list", "--help"]) - self.assertEqual(result.exit_code, 0) + result = runner.invoke(app, [*BASE, "batch", "restart", "al-123", "--yes"]) - def test_batch_results_exists(self) -> None: - result = runner.invoke(app, ["batch", "results", "--help"]) - self.assertEqual(result.exit_code, 0) + self.assertEqual(result.exit_code, 0, result.output) + mock_restart.assert_called_once_with("private-key", "workspace-1", "al-123") if __name__ == "__main__": diff --git a/tests/cli/test_completion_handler.py b/tests/cli/test_completion_handler.py index 41a57ed6..d701543e 100644 --- a/tests/cli/test_completion_handler.py +++ b/tests/cli/test_completion_handler.py @@ -95,7 +95,6 @@ def test_hidden_commands_filtered_from_completion(self) -> None: "get_workspace_info", "run_video_inference_api", "help", - "batch", } leaked = hidden_examples & visible self.assertFalse(leaked, f"Hidden commands leaked into completion: {leaked}")