From 4531c50d76ab9c667e6a842498edaea05f9e4941 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o?= Date: Wed, 4 Nov 2026 23:23:41 -0500 Subject: [PATCH 1/4] feat(workflows): add deploy gates to the SDK and CLI A workflow with a deploy gate no longer publishes through updateWorkflow, so add a client for the gates API (save a version, run its checks, deploy it, manage the Eval gate) and raise a typed error with the server's next steps when updateWorkflow is refused. --- CHANGELOG.md | 21 + CLI-COMMANDS.md | 19 + docs/core/workspace.md | 34 ++ roboflow/adapters/rfapi.py | 17 +- roboflow/adapters/workflowdeploygatesapi.py | 113 +++++ roboflow/adapters/workflowevalsapi.py | 14 +- roboflow/cli/handlers/workflow.py | 527 +++++++++++++++++++- roboflow/core/workflow_deploy_gates.py | 263 ++++++++++ roboflow/core/workspace.py | 20 + tests/adapters/test_rfapi_phase2.py | 48 +- tests/cli/test_workflow_handler.py | 134 ++++- tests/test_workflow_deploy_gates.py | 185 +++++++ 12 files changed, 1365 insertions(+), 30 deletions(-) create mode 100644 roboflow/adapters/workflowdeploygatesapi.py create mode 100644 roboflow/core/workflow_deploy_gates.py create mode 100644 tests/test_workflow_deploy_gates.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3de8e4fd..36699acc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,18 @@ All notable changes to this project will be documented in this file. ### Added +- `Workspace.workflow_deploy_gates()` returns a client for workflow deploy gates. + `save_version` saves a config as a version without publishing it, `run` starts + the version's deploy checks, `get_version` and `wait_for_version` read the + verdicts, and `deploy` publishes a version once it passed every gate. The + Workflow Eval gate is managed with `get_eval_gate`, `set_eval_gate`, + `update_eval_gate`, `set_eval_gate_new_check_thresholds`, `remove_eval_gate` + and `get_eval_checks`. Errors raise `WorkflowDeployGateError`, whose + `next_steps` lists what publishes the workflow. +- `roboflow workflow version save`, `roboflow workflow deploy-gates list|run|status`, + `roboflow workflow deploy-gates eval get|set|update|new-check-thresholds|remove|checks` + and `roboflow workflow deploy`, which replaces the hidden stub. `run --wait` and + `status --wait` exit with code 1 when the version is not publishable. - `Project.autolabel(..., keypoint_confidence_threshold=0.75)` and `roboflow autolabel start --keypoint-confidence 0.75` set the lowest confidence a predicted keypoint needs to be saved as visible when a keypoint-detection @@ -21,6 +33,15 @@ All notable changes to this project will be documented in this file. are already labeled aren't duplicated ([#570](https://github.com/roboflow/roboflow-python/pull/570)). +### Changed + +- `rfapi.update_workflow` and `roboflow workflow update` on a workflow with deploy + gates raise `WorkflowDeployGateError` (a `RoboflowError`) with the server's + `code`, `gate_types` and `next_steps`, instead of a `RoboflowError` holding the + raw JSON. The server saves nothing in that case: save a version, run its + deploy checks and deploy it. Other `update_workflow` and `get_workflow` errors + now carry `status_code`. + ### Deprecated - `Project.autolabel(..., preserve_existing_annotations=True)` still works but diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 5359365c..98a3c158 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -393,6 +393,25 @@ roboflow workflow version list my-workflow roboflow workflow fork other-ws/their-workflow ``` +### Publish a workflow with deploy gates + +A workflow with a deploy gate publishes only a saved version that passed it, so +`workflow update` is refused there. Save a version, run its checks, then deploy it: + +```bash +roboflow workflow version save my-workflow --definition updated.json # prints the version id +roboflow workflow deploy-gates run my-workflow --wait # exits 1 unless publishable +roboflow workflow deploy my-workflow + +# Gate a workflow on a Workflow Eval, with per-check minimum pass rates. +roboflow workflow deploy-gates eval set my-workflow --eval --threshold code-reading=0.98 +roboflow workflow deploy-gates eval get my-workflow +roboflow workflow deploy-gates list +``` + +On an Auto-Publish workflow a version that passes publishes itself, so `deploy` is only needed on +a Protected one. + ### Fork a Universe project (async) ```bash diff --git a/docs/core/workspace.md b/docs/core/workspace.md index a70d9c45..ac0e2629 100644 --- a/docs/core/workspace.md +++ b/docs/core/workspace.md @@ -25,3 +25,37 @@ Pages include Dataset revisions, coverage counts and a possibly unknown (`None`) `total`. Follow `nextCursor` even when `items` is empty, keeping the same query and limit (1–48, default 24). On HTTP 409, restart without a cursor. Preview URLs expire. This search does not execute a Workflow or create embeddings. + +## Publish a Workflow with Deploy Gates + +A workflow with a deploy gate publishes only a saved version that passed every gate, so +`update_workflow` raises `WorkflowDeployGateError` (`code == "deploy_check_required"`) and saves +nothing. Save a version, run its gates and deploy it: + +```python +gates = workspace.workflow_deploy_gates() +workflow_id = workspace.get_workflow("my-workflow")["workflow"]["id"] + +version_id = gates.save_version(workflow_id, definition)["version"]["id"] +gates.run(workflow_id, version_id) +verdicts = gates.wait_for_version(workflow_id, version_id) +if verdicts["publishable"]: + gates.deploy(workflow_id, version_id) +else: + for gate in verdicts["gates"]: + print(gate["type"], gate["status"], gate["reason"]) +``` + +`save_version` and `run` send an `Idempotency-Key`; pass `idempotency_key=` to retry one safely. +On an Auto-Publish workflow a version that passes publishes itself, so `deploy` is only needed on a +Protected one. Scopes: `workflow:update` to save and deploy; `workflow:update`, +`workflow-evals:run`, `workflow:read` and `model:infer` to run; `workflow:read` to read verdicts. +Add `workflow-evals:read` to see a check's Eval Run results. Folder-scoped keys are refused. + +To gate a workflow on a Workflow Eval: + +```python +gates.set_eval_gate(workflow_id, "EVAL_ID", check_thresholds={"code-reading": 0.98}) +``` + +Checks left out need every applicable Case to pass. diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index a381f618..0f957b14 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -1756,7 +1756,7 @@ def get_workflow(api_key, workspace_url, workflow_url): params={"api_key": api_key}, ) if response.status_code != 200: - raise RoboflowError(response.text) + raise RoboflowError(response.text, status_code=response.status_code) return response.json() @@ -1813,6 +1813,14 @@ def update_workflow(api_key, workspace_url, *, workflow_id, workflow_name, workf definitions (``{"version", "inputs", "steps", "outputs"}``) are auto-wrapped in ``{"specification": ...}`` to match the backend's stored shape; see ``_normalize_workflow_config``. + + Raises: + WorkflowDeployGateError: the workflow has deploy gates, so its config + cannot publish this way (``409 deploy_check_required``). Nothing was + saved. ``exc.next_steps`` lists the requests that publish it: save a + version, run its deploy checks and deploy it once they pass (see + ``Workspace.workflow_deploy_gates``). It is a ``RoboflowError``. + RoboflowError: any other failure, with ``status_code`` set. """ config = _normalize_workflow_config(config) payload: Dict[str, str] = { @@ -1827,7 +1835,12 @@ def update_workflow(api_key, workspace_url, *, workflow_id, workflow_name, workf json=payload, ) if response.status_code != 200: - raise RoboflowError(response.text) + from roboflow.adapters.workflowdeploygatesapi import deploy_gate_refusal + + refusal = deploy_gate_refusal(response) + if refusal is not None: + raise refusal + raise RoboflowError(response.text, status_code=response.status_code) return response.json() diff --git a/roboflow/adapters/workflowdeploygatesapi.py b/roboflow/adapters/workflowdeploygatesapi.py new file mode 100644 index 00000000..8714c8d3 --- /dev/null +++ b/roboflow/adapters/workflowdeploygatesapi.py @@ -0,0 +1,113 @@ +"""Transport for the public Workflow deploy gates API. + +Every route lives under ``{API_URL}/workspaces/{workspace}/workflows``. The API +shares the Workflow Evals API's bearer auth, ``Idempotency-Key`` rule and error +envelope, so this module reuses that transport's helpers. The per-endpoint +surface lives in :class:`roboflow.core.workflow_deploy_gates.WorkflowDeployGates`. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, Dict, List, Optional +from urllib.parse import quote + +from roboflow.adapters.workflowevalsapi import ( + DEFAULT_TIMEOUT, + WorkflowEvalError, + _clean_params, + _headers, + error_from_response, +) +from roboflow.config import API_URL +from roboflow.util import http + +if TYPE_CHECKING: + import requests + +# The codes a deploy gate refusal carries, on the deploy gates API and on the +# legacy publish paths (``POST /{workspace}/updateWorkflow``). +DEPLOY_GATE_ERROR_CODES = frozenset( + { + "deploy_check_required", + "deploy_check_not_passed", + "deploy_check_conflict", + "deploy_check_unavailable", + "deploy_check_local_inference_not_allowed", + "deploy_gate_invalid_eval", + "deploy_gate_feature_unavailable", + "deploy_gate_permission_required", + } +) + + +class WorkflowDeployGateError(WorkflowEvalError): + """Error returned by the deploy gates API, or a publish a deploy gate refused. + + Same envelope as :class:`WorkflowEvalError`: ``code``, ``category``, + ``retryable``, ``request_id``, ``message`` and ``details``. A refused publish + (``deploy_check_required`` from ``updateWorkflow``, ``deploy_check_not_passed`` + from :meth:`WorkflowDeployGates.deploy`) lists the requests that publish the + workflow in :attr:`next_steps`. + """ + + @property + def next_steps(self) -> List[Dict[str, Any]]: + """``[{action, method, path, description}]`` that move the workflow towards publishing.""" + steps = self.details.get("nextSteps") + return steps if isinstance(steps, list) else [] + + @property + def gate_types(self) -> List[str]: + """The gate types behind the refusal, e.g. ``["workflow_eval"]``.""" + gate_types = self.details.get("gateTypes") + if isinstance(gate_types, list): + return gate_types + gate_type = self.details.get("gateType") + return [gate_type] if isinstance(gate_type, str) else [] + + +def deploy_gate_refusal(response: requests.Response) -> Optional[WorkflowDeployGateError]: + """Return the deploy gate refusal a non-2xx legacy response carries, else ``None``.""" + error = error_from_response(response, WorkflowDeployGateError) + if isinstance(error, WorkflowDeployGateError) and error.code in DEPLOY_GATE_ERROR_CODES: + return error + return None + + +def base_url(workspace_url: str) -> str: + return f"{API_URL}/workspaces/{quote(workspace_url, safe='')}/workflows" + + +def request( + api_key: str, + workspace_url: str, + method: str, + path: str, + *, + params: Optional[Dict[str, Any]] = None, + body: Any = None, + idempotency_key: Optional[str] = None, + timeout: float = DEFAULT_TIMEOUT, +) -> Any: + """Call one deploy gates route and return the decoded JSON body. + + Raises :class:`WorkflowDeployGateError` for any non-2xx response. + """ + response = http.request( + method.upper(), + f"{base_url(workspace_url)}{path}", + params=_clean_params(params), + json=body, + headers=_headers(api_key, idempotency_key=idempotency_key), + timeout=timeout, + ) + if not 200 <= response.status_code < 300: + raise error_from_response(response, WorkflowDeployGateError) + if response.status_code == 204 or not response.content: + return None + try: + return response.json() + except ValueError: + raise WorkflowDeployGateError( + f"Expected JSON from {method.upper()} {path}", status_code=response.status_code + ) from None diff --git a/roboflow/adapters/workflowevalsapi.py b/roboflow/adapters/workflowevalsapi.py index bec56ad9..4a6206dd 100644 --- a/roboflow/adapters/workflowevalsapi.py +++ b/roboflow/adapters/workflowevalsapi.py @@ -12,7 +12,7 @@ import json import uuid -from typing import TYPE_CHECKING, Any, Dict, Iterator, Optional +from typing import TYPE_CHECKING, Any, Dict, Iterator, Optional, Type from urllib.parse import quote from roboflow.adapters.rfapi import RoboflowError @@ -78,8 +78,10 @@ def to_dict(self) -> Dict[str, Any]: return payload -def error_from_response(response: requests.Response) -> WorkflowEvalError: - """Translate a non-2xx response into a :class:`WorkflowEvalError`.""" +def error_from_response( + response: requests.Response, error_class: Type[WorkflowEvalError] = WorkflowEvalError +) -> WorkflowEvalError: + """Translate a non-2xx response into a :class:`WorkflowEvalError` (or ``error_class``).""" body: Any = None try: body = response.json() @@ -87,7 +89,7 @@ def error_from_response(response: requests.Response) -> WorkflowEvalError: pass error = body.get("error") if isinstance(body, dict) else None if isinstance(error, dict): - return WorkflowEvalError( + return error_class( str(error.get("message") or error.get("code") or response.reason or "Request failed"), status_code=response.status_code, code=error.get("code"), @@ -101,9 +103,9 @@ def error_from_response(response: requests.Response) -> WorkflowEvalError: ) if isinstance(error, str): message = body.get("message") or error - return WorkflowEvalError(str(message), status_code=response.status_code, code=error, body=body) + return error_class(str(message), status_code=response.status_code, code=error, body=body) text = (response.text or "").strip() or response.reason or "Request failed" - return WorkflowEvalError(text, status_code=response.status_code, body=body) + return error_class(text, status_code=response.status_code, body=body) def new_idempotency_key() -> str: diff --git a/roboflow/cli/handlers/workflow.py b/roboflow/cli/handlers/workflow.py index 8db60b67..d4df6050 100644 --- a/roboflow/cli/handlers/workflow.py +++ b/roboflow/cli/handlers/workflow.py @@ -2,7 +2,7 @@ from __future__ import annotations -from typing import Annotated, Optional +from typing import Annotated, Any, Callable, Dict, List, Optional import typer @@ -17,6 +17,43 @@ _version_app = typer.Typer(cls=SortedGroup, help="Manage workflow versions", no_args_is_help=True) workflow_app.add_typer(_version_app, name="version") +# --------------------------------------------------------------------------- +# Sub-apps for ``workflow deploy-gates`` subcommands +# --------------------------------------------------------------------------- + +_deploy_gates_app = typer.Typer( + cls=SortedGroup, + help="Deploy gates: checks a saved version must pass before it publishes", + no_args_is_help=True, +) +workflow_app.add_typer(_deploy_gates_app, name="deploy-gates") + +_eval_gate_app = typer.Typer(cls=SortedGroup, help="Gate a workflow on a Workflow Eval", no_args_is_help=True) +_deploy_gates_app.add_typer(_eval_gate_app, name="eval") + +WorkflowArg = Annotated[ + str, typer.Argument(help="Workflow URL or ID (resolved with a GET, which needs the workflow:read scope)") +] +# Named to sort after workflow_url: SortedGroup orders positional arguments alphabetically. +VersionArg = Annotated[str, typer.Argument(help="Workflow version ID")] +ThresholdOpt = Annotated[ + Optional[List[str]], + typer.Option("--threshold", help="CHECK_ID=RATE minimum pass rate, 0 < RATE <= 1 (repeatable)"), +] +ExpectedUpdatedAtOpt = Annotated[ + Optional[str], + typer.Option( + "--expected-updated-at", + help="The gate's updatedAt as you read it ('null' if it had none); refuses the write if it changed since", + ), +] +WaitOpt = Annotated[bool, typer.Option("--wait", help="Poll until no gate is pending; exit 1 if not publishable")] +TimeoutOpt = Annotated[float, typer.Option("--timeout", help="Seconds to wait before giving up")] +IntervalOpt = Annotated[float, typer.Option("--interval", help="Seconds between polls")] +IdempotencyOpt = Annotated[ + Optional[str], typer.Option("--idempotency-key", help="UUID v4 to reuse when retrying (default: generated)") +] + @workflow_app.command("list") def list_workflows(ctx: typer.Context) -> None: @@ -68,6 +105,21 @@ def list_workflow_versions( _list_workflow_versions(args) +@_version_app.command("save") +def save_workflow_version( + ctx: typer.Context, + workflow_url: WorkflowArg, + definition: Annotated[str, typer.Option("--definition", help="Path to JSON definition file")], + name: Annotated[Optional[str], typer.Option("--name", help="Version name")] = None, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Save a definition as a new version without publishing it.""" + args = ctx_to_args( + ctx, workflow_url=workflow_url, definition=definition, name=name, idempotency_key=idempotency_key + ) + _save_workflow_version(args) + + @workflow_app.command("fork") def fork_workflow( ctx: typer.Context, @@ -99,14 +151,154 @@ def run_workflow( _stub_run(args) -@workflow_app.command("deploy", hidden=True) -def deploy_workflow( +@workflow_app.command("deploy") +def deploy_workflow(ctx: typer.Context, workflow_url: WorkflowArg, workflow_version: VersionArg) -> None: + """Publish a saved version. A gated workflow publishes only a version that passed every gate.""" + args = ctx_to_args(ctx, workflow_url=workflow_url, version_id=workflow_version) + _deploy_workflow_version(args) + + +@_deploy_gates_app.command("list") +def list_deploy_gates( ctx: typer.Context, - workflow_url: Annotated[str, typer.Argument(help="Workflow URL or ID")], + workflow_url: Annotated[ + Optional[str], typer.Argument(help="Workflow URL or ID; omit to list the whole workspace's gates") + ] = None, + gate_type: Annotated[Optional[str], typer.Option("--type", help="Only gates of this type")] = None, + eval_id: Annotated[Optional[str], typer.Option("--eval", help="Only gates backed by this Eval")] = None, +) -> None: + """List deploy gates, of one workflow or of the whole workspace.""" + args = ctx_to_args(ctx, workflow_url=workflow_url, gate_type=gate_type, eval_id=eval_id) + _list_deploy_gates(args) + + +@_deploy_gates_app.command("run") +def run_deploy_gates( + ctx: typer.Context, + workflow_url: WorkflowArg, + workflow_version: VersionArg, + wait: WaitOpt = False, + timeout: TimeoutOpt = 3600, + interval: IntervalOpt = 5, + idempotency_key: IdempotencyOpt = None, +) -> None: + """Start every deploy gate on a version (an Eval gate runs the Eval, which may consume credits).""" + args = ctx_to_args( + ctx, + workflow_url=workflow_url, + version_id=workflow_version, + wait=wait, + timeout=timeout, + interval=interval, + idempotency_key=idempotency_key, + ) + _run_deploy_gates(args) + + +@_deploy_gates_app.command("status") +def deploy_gates_status( + ctx: typer.Context, + workflow_url: WorkflowArg, + workflow_version: VersionArg, + wait: WaitOpt = False, + timeout: TimeoutOpt = 3600, + interval: IntervalOpt = 5, +) -> None: + """Show every gate's verdict on a version and whether it is publishable.""" + args = ctx_to_args( + ctx, workflow_url=workflow_url, version_id=workflow_version, wait=wait, timeout=timeout, interval=interval + ) + _deploy_gates_status(args) + + +@_eval_gate_app.command("get") +def get_eval_gate(ctx: typer.Context, workflow_url: WorkflowArg) -> None: + """Show the Eval gate with the Eval's checks and their thresholds.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + _gate_call(args, lambda gates, wf: gates.get_eval_gate(wf), _eval_gate_text) + + +@_eval_gate_app.command("set") +def set_eval_gate( + ctx: typer.Context, + workflow_url: WorkflowArg, + eval_id: Annotated[str, typer.Option("--eval", help="Eval ID")], + threshold: ThresholdOpt = None, + expected_updated_at: ExpectedUpdatedAtOpt = None, +) -> None: + """Gate a workflow on an Eval; checks without --threshold need every applicable Case to pass.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + thresholds = _parse_thresholds(args, threshold) + expected = _expected_updated_at(expected_updated_at) + _gate_call( + args, + lambda gates, wf: gates.set_eval_gate(wf, eval_id, check_thresholds=thresholds, expected_updated_at=expected), + _eval_gate_text, + ) + + +@_eval_gate_app.command("update") +def update_eval_gate( + ctx: typer.Context, + workflow_url: WorkflowArg, + threshold: Annotated[ + List[str], typer.Option("--threshold", help="CHECK_ID=RATE minimum pass rate, 0 < RATE <= 1 (repeatable)") + ], + expected_updated_at: ExpectedUpdatedAtOpt = None, +) -> None: + """Change some checks' minimum pass rates; the others keep theirs.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + thresholds = _parse_thresholds(args, threshold) + expected = _expected_updated_at(expected_updated_at) + _gate_call( + args, + lambda gates, wf: gates.update_eval_gate(wf, thresholds, expected_updated_at=expected), + _eval_gate_text, + ) + + +@_eval_gate_app.command("new-check-thresholds") +def set_eval_gate_new_check_thresholds( + ctx: typer.Context, + workflow_url: WorkflowArg, + threshold: Annotated[ + List[str], typer.Option("--threshold", help="CHECK_ID=RATE minimum pass rate, 0 < RATE <= 1 (repeatable)") + ], ) -> None: - """Deploy a workflow.""" + """Set thresholds for checks added to the Eval after the gate was set.""" args = ctx_to_args(ctx, workflow_url=workflow_url) - _stub_deploy(args) + thresholds = _parse_thresholds(args, threshold) + _gate_call(args, lambda gates, wf: gates.set_eval_gate_new_check_thresholds(wf, thresholds), _eval_gate_text) + + +@_eval_gate_app.command("remove") +def remove_eval_gate( + ctx: typer.Context, + workflow_url: WorkflowArg, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt.")] = False, +) -> None: + """Remove the Eval gate, so the workflow publishes without passing it.""" + from roboflow.cli._output import confirm_destructive + + args = ctx_to_args(ctx, workflow_url=workflow_url, yes=yes) + if not confirm_destructive(args, f"Remove the Eval deploy gate from '{workflow_url}'?"): + return + _gate_call( + args, + lambda gates, wf: gates.remove_eval_gate(wf), + lambda _status: f"Removed the Eval deploy gate from {workflow_url}.", + ) + + +@_eval_gate_app.command("checks") +def list_eval_gate_checks( + ctx: typer.Context, + workflow_url: WorkflowArg, + version_id: Annotated[Optional[str], typer.Option("--version", help="Only this version's check")] = None, +) -> None: + """Show the Eval deploy checks of every version, or of one.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + _gate_call(args, lambda gates, wf: gates.get_eval_checks(wf, version_id=version_id), _eval_checks_text) @workflow_app.command("delete") @@ -270,6 +462,7 @@ def _update_workflow(args) -> None: # noqa: ANN001 import json as _json from roboflow.adapters import rfapi + from roboflow.adapters.workflowdeploygatesapi import WorkflowDeployGateError from roboflow.cli._output import output, output_error resolved = _resolve_workspace_and_key(args) @@ -314,6 +507,9 @@ def _update_workflow(args) -> None: # noqa: ANN001 workflow_url=workflow_url_slug, config=config, ) + except WorkflowDeployGateError as exc: + _fail_gate(args, exc) + return except rfapi.RoboflowError as exc: output_error(args, str(exc)) return @@ -416,16 +612,6 @@ def _stub_run(args) -> None: # noqa: ANN001 ) -def _stub_deploy(args) -> None: # noqa: ANN001 - from roboflow.cli._output import output_error - - output_error( - args, - "This command is not yet implemented.", - hint="Coming in a future release.", - ) - - def _delete_workflow(args) -> None: # noqa: ANN001 from roboflow.adapters import rfapi from roboflow.cli._output import confirm_destructive, output, output_api_error @@ -537,3 +723,312 @@ def _restore_workflow(args) -> None: # noqa: ANN001 return output(args, data, text=f"Restored {workspace_url}/{args.workflow_url} from Trash.") + + +# --------------------------------------------------------------------------- +# Deploy gates +# --------------------------------------------------------------------------- + + +def _resolve_workflow_id(args, api_key: str, workspace_url: str) -> Optional[str]: # noqa: ANN001 + """Return the workflow document id for ``args.workflow_url`` (a URL slug or an id).""" + from roboflow.adapters import rfapi + from roboflow.cli._output import output_api_error, output_error + + try: + data = rfapi.get_workflow(api_key, workspace_url, args.workflow_url) + except rfapi.RoboflowError as exc: + output_api_error( + args, + exc, + hint="Reading the workflow needs the workflow:read scope.", + not_found_hint="Run 'roboflow workflow list' to see the workspace's workflows.", + ) + return None + workflow = data.get("workflow", data) if isinstance(data, dict) else None + workflow_id = workflow.get("id") if isinstance(workflow, dict) else None + if not workflow_id: + output_error(args, "Unexpected response from API when fetching workflow.") + return None + return str(workflow_id) + + +def _gate_hint(args, exc) -> Optional[str]: # noqa: ANN001 + workflow = getattr(args, "workflow_url", None) or "" + version = getattr(args, "version_id", None) or "" + definition = getattr(args, "definition", None) or "" + hints = { + "deploy_check_required": ( + "This workflow has deploy gates, so it publishes only a saved version that passed them. Run: " + f"roboflow workflow version save {workflow} --definition {definition}, then " + f"roboflow workflow deploy-gates run {workflow} --wait, then " + f"roboflow workflow deploy {workflow} ." + ), + "deploy_check_not_passed": ( + f"Run the version's deploy checks first: roboflow workflow deploy-gates run {workflow} {version} --wait." + ), + "deploy_gate_feature_unavailable": "Deploy gates need a plan with workflow version history.", + "deploy_gate_permission_required": "Changing deploy gates needs the Manage Workflow Deploy Gates permission.", + "deploy_gate_invalid_eval": "The Eval cannot gate this workflow; see error.details.reason.", + "deploy_check_local_inference_not_allowed": "Deploy checks run on the server. Set the Eval to run there.", + "deploy_check_unavailable": "Deploy checks cannot start right now. Try again later.", + } + if exc.code in hints: + return hints[exc.code] + if exc.status_code == 401: + return "Check your API key, or run 'roboflow auth login'." + if exc.status_code == 403: + return ( + "The API key needs workflow:read or workflow:update (run also needs workflow-evals:run and model:infer) " + "and cannot be folder-scoped." + ) + return None + + +def _fail_gate(args, exc) -> None: # noqa: ANN001 + """Render a deploy gates API error as a structured CLI error and exit.""" + import json + + from roboflow.cli._output import output_error + + payload = exc.to_dict() + payload.pop("hint", None) + if exc.status_code in (401, 403): + exit_code = 2 + elif exc.status_code == 404: + exit_code = 3 + else: + exit_code = 1 + output_error(args, json.dumps({"error": payload}), hint=_gate_hint(args, exc), exit_code=exit_code) + + +def _gate_call( + args, # noqa: ANN001 + call: Callable[[Any, Optional[str]], Any], + text: Optional[Callable[[Any], str]] = None, + *, + needs_workflow: bool = True, +) -> Any: + """Resolve the client (and workflow id), run *call*, and print its result. Returns the result.""" + from roboflow.adapters.workflowevalsapi import WorkflowEvalError + from roboflow.cli._output import output, output_error, suppress_sdk_output + from roboflow.core.workflow_deploy_gates import WorkflowDeployGates + + resolved = _resolve_workspace_and_key(args) + if resolved is None: + return None + workspace_url, api_key = resolved + workflow_id = _resolve_workflow_id(args, api_key, workspace_url) if needs_workflow else None + if needs_workflow and workflow_id is None: + return None + try: + with suppress_sdk_output(args): + result = call(WorkflowDeployGates(api_key, workspace_url), workflow_id) + except WorkflowEvalError as exc: + _fail_gate(args, exc) + return None + except TimeoutError as exc: + output_error( + args, + str(exc), + hint="The checks keep running. Run 'roboflow workflow deploy-gates status ... --wait' to keep waiting.", + ) + return None + output(args, result, text=text(result) if text and not getattr(args, "json", False) else None) + return result + + +def _parse_thresholds(args, values: Optional[List[str]]) -> Optional[Dict[str, float]]: # noqa: ANN001 + from roboflow.cli._output import output_error + + if not values: + return None + thresholds: Dict[str, float] = {} + for value in values: + check_id, _, rate = value.rpartition("=") + try: + parsed: Optional[float] = float(rate) + except ValueError: + parsed = None + if not check_id or parsed is None: + output_error( + args, + f"Invalid --threshold '{value}'.", + hint="Use CHECK_ID=RATE with 0 < RATE <= 1, e.g. --threshold code-reading=0.98.", + ) + return None + thresholds[check_id] = parsed + return thresholds + + +def _expected_updated_at(value: Optional[str]) -> Any: + from roboflow.core.workflow_deploy_gates import UNSET + + if value is None: + return UNSET + return None if value.lower() == "null" else value + + +def _wait_for(args, gates, workflow_id: Optional[str]) -> Dict[str, Any]: # noqa: ANN001 + import sys + + if not args.json and not getattr(args, "quiet", False): + print(f"Waiting for the deploy gates on version {args.version_id}...", file=sys.stderr) + return gates.wait_for_version(workflow_id, args.version_id, timeout=args.timeout, interval=args.interval) + + +def _exit_unless_publishable(args, verdicts: Any) -> None: # noqa: ANN001 + import sys + + if args.wait and isinstance(verdicts, dict) and not verdicts.get("publishable"): + sys.exit(1) + + +def _verdicts_text(verdicts: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + standing = "publishable" if verdicts.get("publishable") else "not publishable" + lines = [f"Version {verdicts.get('versionId', '')}: {standing}"] + gates = verdicts.get("gates") or [] + if gates: + rows = [ + {"type": g.get("type", ""), "status": g.get("status", ""), "reason": g.get("reason") or ""} for g in gates + ] + lines += ["", format_table(rows, columns=["type", "status", "reason"], headers=["GATE", "STATUS", "REASON"])] + else: + lines.append("The workflow has no deploy gates.") + return "\n".join(lines) + + +def _rate(rate: Any) -> str: + return f"{rate:.0%}" if isinstance(rate, (int, float)) else "not set (100%)" + + +def _eval_gate_text(status: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + if not status.get("gate"): + return "No Eval deploy gate." + evaluation = status.get("eval") or {} + gate = status.get("gate") or {} + lines = [ + f"Eval: {evaluation.get('name') or gate.get('evalId', '')} ({gate.get('evalId', '')})", + f"Updated: {gate.get('updatedAt', '')}", + ] + checks = status.get("checks") or [] + if checks: + rows = [ + {"id": c.get("id", ""), "name": c.get("name", ""), "rate": _rate(c.get("minimumPassRate"))} for c in checks + ] + lines += ["", format_table(rows, columns=["id", "name", "rate"], headers=["CHECK", "NAME", "MIN PASS RATE"])] + return "\n".join(lines) + + +def _eval_checks_text(data: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + items = data.get("items") if "items" in data else [data] + rows = [ + { + "version": c.get("versionId", ""), + "status": c.get("status", ""), + "current": "yes" if c.get("current") else f"no ({c.get('staleReason') or 'stale'})", + "started": c.get("startedAt", ""), + } + for c in items or [] + ] + return format_table( + rows, columns=["version", "status", "current", "started"], headers=["VERSION", "STATUS", "CURRENT", "STARTED"] + ) + + +def _workspace_gates_text(data: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + rows = [ + { + "workflow": g.get("workflowUrl") or g.get("workflowId", ""), + "type": g.get("type", ""), + "eval": g.get("evalId", ""), + } + for g in data.get("items") or [] + ] + return format_table(rows, columns=["workflow", "type", "eval"], headers=["WORKFLOW", "GATE", "EVAL"]) + + +def _workflow_gates_text(data: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + rows = [] + for item in data.get("items") or []: + status = item.get("status") or {} + evaluation = status.get("eval") or {} + rows.append( + { + "type": item.get("type", ""), + "eval": evaluation.get("name") or (status.get("gate") or {}).get("evalId", ""), + } + ) + if not rows: + return "The workflow has no deploy gates." + return format_table(rows, columns=["type", "eval"], headers=["GATE", "EVAL"]) + + +def _save_workflow_version(args) -> None: # noqa: ANN001 + definition = _read_definition_file(args) + if definition is False: + return + + def text(saved: Dict[str, Any]) -> str: + version = saved.get("version") or {} + version_id = version.get("id", "") + if saved.get("replayed"): + head = f"Version {version_id} was already saved with this idempotency key. It is not published." + else: + head = f"Saved version {version_id} of {args.workflow_url}. It is not published." + return ( + f"{head}\nRun its deploy checks: roboflow workflow deploy-gates run {args.workflow_url} {version_id} --wait" + ) + + _gate_call( + args, + lambda gates, wf: gates.save_version( + wf, definition, version_name=args.name, idempotency_key=args.idempotency_key + ), + text, + ) + + +def _deploy_workflow_version(args) -> None: # noqa: ANN001 + def text(deployed: Dict[str, Any]) -> str: + version_id = deployed.get("versionId", args.version_id) + return f"Published version {version_id} of {deployed.get('workflowUrl', args.workflow_url)}." + + _gate_call(args, lambda gates, wf: gates.deploy(wf, args.version_id), text) + + +def _list_deploy_gates(args) -> None: # noqa: ANN001 + if args.workflow_url: + _gate_call(args, lambda gates, wf: gates.list_for_workflow(wf), _workflow_gates_text) + else: + _gate_call( + args, + lambda gates, _wf: gates.list(gate_type=args.gate_type, eval_id=args.eval_id), + _workspace_gates_text, + needs_workflow=False, + ) + + +def _run_deploy_gates(args) -> None: # noqa: ANN001 + def call(gates, workflow_id: Optional[str]) -> Any: # noqa: ANN001 + started = gates.run(workflow_id, args.version_id, idempotency_key=args.idempotency_key) + return _wait_for(args, gates, workflow_id) if args.wait else started + + _exit_unless_publishable(args, _gate_call(args, call, _verdicts_text)) + + +def _deploy_gates_status(args) -> None: # noqa: ANN001 + def call(gates, workflow_id: Optional[str]) -> Any: # noqa: ANN001 + return _wait_for(args, gates, workflow_id) if args.wait else gates.get_version(workflow_id, args.version_id) + + _exit_unless_publishable(args, _gate_call(args, call, _verdicts_text)) diff --git a/roboflow/core/workflow_deploy_gates.py b/roboflow/core/workflow_deploy_gates.py new file mode 100644 index 00000000..051fac48 --- /dev/null +++ b/roboflow/core/workflow_deploy_gates.py @@ -0,0 +1,263 @@ +"""Workflow deploy gates: wraps the public ``/workspaces/{workspace}/workflows`` gate API. + +A deploy gate is a condition a saved Workflow version must meet before it +publishes. Today the only type is ``workflow_eval``: a fresh Run of the version +on one Workflow Eval must reach every check's minimum pass rate. While a +workflow has a gate, ``updateWorkflow`` no longer publishes its config; a +version publishes only through :meth:`WorkflowDeployGates.deploy`, once it +passed every gate. Get a client from +:meth:`roboflow.core.workspace.Workspace.workflow_deploy_gates`:: + + import roboflow + + gates = roboflow.Roboflow().workspace().workflow_deploy_gates() + saved = gates.save_version(workflow_id, definition) + version_id = saved["version"]["id"] + gates.run(workflow_id, version_id) + verdicts = gates.wait_for_version(workflow_id, version_id) + if verdicts["publishable"]: + gates.deploy(workflow_id, version_id) + +On an Auto-Publish workflow a version that passed every gate publishes itself +while it is still the latest saved draft, so :meth:`deploy` is only needed on a +Protected one. + +Conventions shared by every method: + +* Methods take the workflow document id (``workflow["id"]`` from + :meth:`Workspace.get_workflow`), not its URL slug. +* :meth:`save_version` and :meth:`run` send an ``Idempotency-Key``. One is + generated when ``idempotency_key`` is omitted; pass your own to safely retry + a request whose response was lost. +* Methods return the server's JSON as plain dicts. +* Errors raise :class:`roboflow.adapters.workflowdeploygatesapi.WorkflowDeployGateError`. + +Scopes: reads need ``workflow:read``, writes ``workflow:update``. :meth:`run` +also needs ``workflow-evals:run`` and ``model:infer``, and +:meth:`set_eval_gate_new_check_thresholds` also ``workflow-evals:write``. A +check's Eval Run results come back only with ``workflow-evals:read``. +Folder-scoped keys are refused. +""" + +from __future__ import annotations + +import time +from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Union + +from roboflow.adapters import workflowdeploygatesapi as api +from roboflow.adapters.rfapi import _normalize_workflow_config +from roboflow.adapters.workflowevalsapi import new_idempotency_key, segment + +ThresholdsArg = Union[Mapping[str, float], Iterable[Dict[str, Any]]] + + +class _Unset: + def __repr__(self) -> str: + return "UNSET" + + +UNSET: Any = _Unset() +"""Default of ``expected_updated_at``: leave the stale-write check out.""" + + +def _thresholds(check_thresholds: ThresholdsArg) -> List[Dict[str, Any]]: + """Accept ``{check_id: rate}`` or the API's ``[{checkId, minimumPassRate}]``.""" + if isinstance(check_thresholds, Mapping): + return [{"checkId": check_id, "minimumPassRate": rate} for check_id, rate in check_thresholds.items()] + return list(check_thresholds) + + +class WorkflowDeployGates: + """Client for one workspace's workflow deploy gates.""" + + def __init__(self, api_key: str, workspace_url: str) -> None: + self._api_key = api_key + self.workspace_url = workspace_url + + # -- transport ---------------------------------------------------------- + + def _call(self, method: str, path: str, **kwargs: Any) -> Any: + return api.request(self._api_key, self.workspace_url, method, path, **kwargs) + + @staticmethod + def _workflow_path(workflow_id: str) -> str: + return f"/{segment(workflow_id)}" + + @classmethod + def _version_path(cls, workflow_id: str, version_id: str) -> str: + return f"{cls._workflow_path(workflow_id)}/versions/{segment(version_id)}" + + @classmethod + def _eval_gate_path(cls, workflow_id: str) -> str: + return f"{cls._workflow_path(workflow_id)}/deploy-gates/workflow-eval" + + # -- discovery ---------------------------------------------------------- + + def openapi(self) -> Dict[str, Any]: + """Return the deploy gates API's OpenAPI document.""" + return self._call("GET", "/deploy-gates/openapi.json") + + def list(self, *, gate_type: Optional[str] = None, eval_id: Optional[str] = None) -> Dict[str, Any]: + """List the workspace's deploy gates, optionally of one type or backed by one Eval.""" + return self._call("GET", "/deploy-gates", params={"type": gate_type, "evalId": eval_id}) + + def list_for_workflow(self, workflow_id: str) -> Dict[str, Any]: + """Return ``{"items": [{type, status}]}``, every gate the workflow has.""" + return self._call("GET", f"{self._workflow_path(workflow_id)}/deploy-gates") + + # -- versions and publishing -------------------------------------------- + + def save_version( + self, + workflow_id: str, + config: Union[str, Dict[str, Any]], + *, + version_name: Optional[str] = None, + idempotency_key: Optional[str] = None, + ) -> Dict[str, Any]: + """Save a config as the workflow's latest version. It never publishes. + + ``config`` is the Workflow JSON as a dict or a JSON string; a bare + definition (``{"version", "inputs", "steps", "outputs"}``) is wrapped as + ``update_workflow`` does. Returns ``{workflowId, version: {id, name, + createdAt, email}, replayed}``; ``replayed`` is true when this + ``idempotency_key`` had already saved the version. + """ + body: Dict[str, Any] = {"config": _normalize_workflow_config(config)} + if version_name is not None: + body["versionName"] = version_name + return self._call( + "POST", + f"{self._workflow_path(workflow_id)}/versions", + body=body, + idempotency_key=idempotency_key or new_idempotency_key(), + ) + + def get_version(self, workflow_id: str, version_id: str) -> Dict[str, Any]: + """Return every gate's verdict on the version: ``{versionId, gates, publishable}``. + + Each gate is ``{type, status, reason}`` with ``status`` one of + ``not_started``, ``pending``, ``passed`` or ``failed``. + """ + return self._call("GET", f"{self._version_path(workflow_id, version_id)}/deploy-gates") + + def run(self, workflow_id: str, version_id: str, *, idempotency_key: Optional[str] = None) -> Dict[str, Any]: + """Start every deploy gate of the workflow on the version and return the verdicts. + + A replay with the same ``idempotency_key`` returns the check it started, + even a failed one, and never starts another Run. A new key reuses a + pending or passed check and starts a new one on a failed, errored, + cancelled or outdated check. A Workflow Eval gate runs the Eval on the + server, which may consume credits. + """ + return self._call( + "POST", + f"{self._version_path(workflow_id, version_id)}/deploy-gates/run", + body={}, + idempotency_key=idempotency_key or new_idempotency_key(), + ) + + def wait_for_version( + self, + workflow_id: str, + version_id: str, + *, + timeout: float = 3600, + interval: float = 5, + on_poll: Optional[Callable[[Dict[str, Any]], None]] = None, + ) -> Dict[str, Any]: + """Poll :meth:`get_version` until no gate is ``pending`` and return the verdicts. + + Check ``publishable`` on the result: a gate can end ``failed``, or + ``not_started`` when nothing was started. Stopping the wait never + cancels a check. Raises ``TimeoutError`` when ``timeout`` seconds elapse + first. + """ + deadline = time.monotonic() + timeout + while True: + verdicts = self.get_version(workflow_id, version_id) + if on_poll: + on_poll(verdicts) + if not any(gate.get("status") == "pending" for gate in verdicts.get("gates") or []): + return verdicts + if time.monotonic() >= deadline: + raise TimeoutError(f"Version {version_id} still has pending deploy gates after {timeout:g}s") + time.sleep(interval) + + def deploy(self, workflow_id: str, version_id: str) -> Dict[str, Any]: + """Publish the version and return ``{workflowId, workflowUrl, versionId, status}``. + + On a gated workflow it is refused with ``409 deploy_check_not_passed`` + until the version passed every gate. Needs a plan with workflow version + history. + """ + return self._call("POST", f"{self._version_path(workflow_id, version_id)}/deploy", body={}) + + # -- workflow_eval gate ------------------------------------------------- + + def get_eval_gate(self, workflow_id: str) -> Dict[str, Any]: + """Return the Eval gate's status: ``{gate, checks, eval, subject}`` (``gate`` is null without one).""" + return self._call("GET", self._eval_gate_path(workflow_id)) + + def set_eval_gate( + self, + workflow_id: str, + eval_id: str, + *, + check_thresholds: Optional[ThresholdsArg] = None, + expected_updated_at: Optional[str] = UNSET, + ) -> Dict[str, Any]: + """Gate the workflow on an Eval, with optional per-check minimum pass rates. + + ``check_thresholds`` is ``{check_id: rate}`` or a list of + ``{"checkId", "minimumPassRate"}``, with ``0 < rate <= 1``; checks left + out need every applicable Case to pass. ``expected_updated_at`` is the + gate's ``updatedAt`` as you read it, or ``None`` when you read no gate: + the write is refused with 409 if the gate changed since. Leave it out + to skip that check. Setting the gate retires earlier checks. + """ + body: Dict[str, Any] = {"evalId": eval_id} + if check_thresholds is not None: + body["checkThresholds"] = _thresholds(check_thresholds) + if expected_updated_at is not UNSET: + body["expectedUpdatedAt"] = expected_updated_at + return self._call("PUT", self._eval_gate_path(workflow_id), body=body) + + def update_eval_gate( + self, + workflow_id: str, + check_thresholds: ThresholdsArg, + *, + expected_updated_at: Optional[str] = UNSET, + ) -> Dict[str, Any]: + """Change some checks' minimum pass rates; the others keep theirs. + + Same ``check_thresholds`` and ``expected_updated_at`` as :meth:`set_eval_gate`. + """ + body: Dict[str, Any] = {"checkThresholds": _thresholds(check_thresholds)} + if expected_updated_at is not UNSET: + body["expectedUpdatedAt"] = expected_updated_at + return self._call("PATCH", self._eval_gate_path(workflow_id), body=body) + + def set_eval_gate_new_check_thresholds(self, workflow_id: str, check_thresholds: ThresholdsArg) -> Dict[str, Any]: + """Set thresholds for checks added to the Eval after the gate was set. + + Only checks without a threshold in the gate yet are accepted; until set + they need every applicable Case to pass. + """ + return self._call( + "POST", + f"{self._eval_gate_path(workflow_id)}/new-check-thresholds", + body={"checkThresholds": _thresholds(check_thresholds)}, + ) + + def remove_eval_gate(self, workflow_id: str) -> Dict[str, Any]: + """Remove the Eval gate and return the gate status after the removal.""" + return self._call("DELETE", self._eval_gate_path(workflow_id)) + + def get_eval_checks(self, workflow_id: str, *, version_id: Optional[str] = None) -> Dict[str, Any]: + """Return one version's Eval deploy check, or ``{"items": [...]}`` for every version. + + With ``version_id`` the server answers 404 until a check was started. + """ + return self._call("GET", f"{self._eval_gate_path(workflow_id)}/checks", params={"versionId": version_id}) diff --git a/roboflow/core/workspace.py b/roboflow/core/workspace.py index 5392b071..3e6c6fa8 100644 --- a/roboflow/core/workspace.py +++ b/roboflow/core/workspace.py @@ -21,6 +21,7 @@ if TYPE_CHECKING: from roboflow.core.device import Device from roboflow.core.model_eval import ModelEval + from roboflow.core.workflow_deploy_gates import WorkflowDeployGates from roboflow.core.workflow_evals import WorkflowEvals @@ -1717,6 +1718,25 @@ def workflow_evals(self) -> "WorkflowEvals": return WorkflowEvals(self.__api_key, self.url) + def workflow_deploy_gates(self) -> "WorkflowDeployGates": + """Return a client for this workspace's workflow deploy gates. + + A workflow with a deploy gate publishes only a saved version that passed + it, so ``updateWorkflow`` refuses to publish its config. Save a version, + run its gates and deploy it instead. + + Example: + >>> gates = rf.workspace().workflow_deploy_gates() + >>> workflow_id = rf.workspace().get_workflow("my-workflow")["workflow"]["id"] + >>> version_id = gates.save_version(workflow_id, definition)["version"]["id"] + >>> gates.run(workflow_id, version_id) + >>> if gates.wait_for_version(workflow_id, version_id)["publishable"]: + ... gates.deploy(workflow_id, version_id) + """ + from roboflow.core.workflow_deploy_gates import WorkflowDeployGates + + return WorkflowDeployGates(self.__api_key, self.url) + def trash(self) -> dict: """ List items currently in the workspace Trash. diff --git a/tests/adapters/test_rfapi_phase2.py b/tests/adapters/test_rfapi_phase2.py index 4d9110ce..b0f573a6 100644 --- a/tests/adapters/test_rfapi_phase2.py +++ b/tests/adapters/test_rfapi_phase2.py @@ -515,8 +515,54 @@ def test_error(self, mock_post): from roboflow.adapters.rfapi import RoboflowError, update_workflow mock_post.return_value = MagicMock(status_code=500, text="Server error") - with self.assertRaises(RoboflowError): + with self.assertRaises(RoboflowError) as ctx: + update_workflow("key", "ws", workflow_id="id-1", workflow_name="WF1", workflow_url="wf1", config="{}") + self.assertEqual(ctx.exception.status_code, 500) + self.assertEqual(str(ctx.exception), "Server error") + + @patch("roboflow.util.http.requests.post") + def test_gated_workflow_raises_the_deploy_gate_refusal(self, mock_post): + from roboflow.adapters.rfapi import RoboflowError, update_workflow + from roboflow.adapters.workflowdeploygatesapi import WorkflowDeployGateError + + next_steps = [ + {"action": "save_version", "method": "POST", "path": "/workspaces/ws/workflows/id-1/versions"}, + { + "action": "deploy_version", + "method": "POST", + "path": "/workspaces/ws/workflows/id-1/versions/{versionId}/deploy", + }, + ] + body = { + "error": { + "message": "This workflow has deploy gates, so updateWorkflow cannot publish its config.", + "status": 409, + "code": "deploy_check_required", + "details": {"gateTypes": ["workflow_eval"], "nextSteps": next_steps}, + } + } + mock_post.return_value = MagicMock(status_code=409, json=lambda: body, text=json.dumps(body)) + with self.assertRaises(WorkflowDeployGateError) as ctx: + update_workflow("key", "ws", workflow_id="id-1", workflow_name="WF1", workflow_url="wf1", config="{}") + error = ctx.exception + self.assertIsInstance(error, RoboflowError) + self.assertEqual(error.status_code, 409) + self.assertEqual(error.code, "deploy_check_required") + self.assertEqual(str(error), body["error"]["message"]) + self.assertEqual(error.gate_types, ["workflow_eval"]) + self.assertEqual(error.next_steps, next_steps) + + @patch("roboflow.util.http.requests.post") + def test_other_409s_stay_plain_errors(self, mock_post): + from roboflow.adapters.rfapi import RoboflowError, update_workflow + from roboflow.adapters.workflowdeploygatesapi import WorkflowDeployGateError + + body = {"error": {"message": "Name taken", "code": "CONFLICT"}} + mock_post.return_value = MagicMock(status_code=409, json=lambda: body, text=json.dumps(body)) + with self.assertRaises(RoboflowError) as ctx: update_workflow("key", "ws", workflow_id="id-1", workflow_name="WF1", workflow_url="wf1", config="{}") + self.assertNotIsInstance(ctx.exception, WorkflowDeployGateError) + self.assertEqual(str(ctx.exception), json.dumps(body)) class TestListWorkflowVersions(unittest.TestCase): diff --git a/tests/cli/test_workflow_handler.py b/tests/cli/test_workflow_handler.py index 128fb377..af5f4487 100644 --- a/tests/cli/test_workflow_handler.py +++ b/tests/cli/test_workflow_handler.py @@ -7,9 +7,11 @@ from argparse import Namespace from unittest.mock import patch +import responses from typer.testing import CliRunner from roboflow.cli import app +from roboflow.config import API_URL runner = CliRunner() @@ -75,6 +77,21 @@ def test_workflow_deploy_exists(self) -> None: result = runner.invoke(app, ["workflow", "deploy", "--help"]) self.assertEqual(result.exit_code, 0) + def test_deploy_gate_commands_exist(self) -> None: + for argv in ( + ["version", "save"], + ["deploy-gates", "list"], + ["deploy-gates", "run"], + ["deploy-gates", "status"], + *( + ["deploy-gates", "eval", c] + for c in ("get", "set", "update", "new-check-thresholds", "remove", "checks") + ), + ): + with self.subTest(command=" ".join(argv)): + result = runner.invoke(app, ["workflow", *argv, "--help"]) + self.assertEqual(result.exit_code, 0, result.output) + class TestWorkflowList(unittest.TestCase): @patch("roboflow.adapters.rfapi.list_workflows") @@ -375,12 +392,119 @@ def test_run_stub(self): with self.assertRaises(SystemExit): _stub_run(args) - def test_deploy_stub(self): - from roboflow.cli.handlers.workflow import _stub_deploy - args = _make_args() - with self.assertRaises(SystemExit): - _stub_deploy(args) +GATES = f"{API_URL}/workspaces/test-ws/workflows/wf-123" +WORKFLOW = {"workflow": {"id": "wf-123", "name": "My WF", "url": "my-wf", "config": "{}"}} + + +def _gates_cli(*argv: str): + return runner.invoke(app, ["--workspace", "test-ws", "--api-key", "test-key", *argv]) + + +def _last_json(output: str): + return json.loads(output.strip().splitlines()[-1]) + + +class TestWorkflowDeployGatesCli(unittest.TestCase): + def setUp(self) -> None: + responses.start() + self.addCleanup(responses.stop) + self.addCleanup(responses.reset) + responses.add(responses.GET, f"{API_URL}/test-ws/workflows/my-wf", json=WORKFLOW, status=200) + + def test_update_on_a_gated_workflow_names_the_commands_that_publish(self) -> None: + refusal = { + "error": { + "message": "This workflow has deploy gates, so updateWorkflow cannot publish its config.", + "status": 409, + "code": "deploy_check_required", + "details": {"gateTypes": ["workflow_eval"], "nextSteps": [{"action": "save_version"}]}, + } + } + responses.add(responses.POST, f"{API_URL}/test-ws/updateWorkflow", json=refusal, status=409) + result = _gates_cli("--json", "workflow", "update", "my-wf") + self.assertEqual(result.exit_code, 1) + error = _last_json(result.output)["error"] + self.assertEqual(error["code"], "deploy_check_required") + self.assertEqual(error["details"]["nextSteps"], [{"action": "save_version"}]) + self.assertIn("roboflow workflow version save my-wf", error["hint"]) + + def test_version_save_resolves_the_slug_and_does_not_publish(self) -> None: + saved = {"workflowId": "wf-123", "version": {"id": "v1", "name": "v1"}, "replayed": False} + responses.add(responses.POST, f"{GATES}/versions", json=saved, status=201) + with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f: + json.dump({"version": "1.0", "inputs": [], "steps": [], "outputs": []}, f) + self.addCleanup(os.unlink, f.name) + result = _gates_cli("workflow", "version", "save", "my-wf", "--definition", f.name) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("not published", result.output) + self.assertIn("roboflow workflow deploy-gates run my-wf v1 --wait", result.output) + self.assertEqual(len(responses.calls), 2) + self.assertTrue(responses.calls[1].request.headers["Idempotency-Key"]) + + def test_run_wait_exits_1_unless_the_version_is_publishable(self) -> None: + for final, publishable, exit_code in (("failed", False, 1), ("passed", True, 0)): + with self.subTest(final=final): + responses.reset() + responses.add(responses.GET, f"{API_URL}/test-ws/workflows/my-wf", json=WORKFLOW, status=200) + pending = {"versionId": "v1", "gates": [{"type": "workflow_eval", "status": "pending"}]} + responses.add(responses.POST, f"{GATES}/versions/v1/deploy-gates/run", json=pending, status=202) + decided = { + "versionId": "v1", + "gates": [{"type": "workflow_eval", "status": final, "reason": None}], + "publishable": publishable, + } + responses.add(responses.GET, f"{GATES}/versions/v1/deploy-gates", json=decided, status=200) + result = _gates_cli("workflow", "deploy-gates", "run", "my-wf", "v1", "--wait", "--interval", "0") + self.assertEqual(result.exit_code, exit_code, result.output) + self.assertIn(final, result.output) + + def test_eval_set_sends_thresholds_and_the_null_stale_write_guard(self) -> None: + responses.add(responses.PUT, f"{GATES}/deploy-gates/workflow-eval", json={"gate": None}, status=200) + result = _gates_cli( + "workflow", + "deploy-gates", + "eval", + "set", + "my-wf", + "--eval", + "e1", + "--threshold", + "code-reading=0.98", + "--expected-updated-at", + "null", + ) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + json.loads(responses.calls[1].request.body), + { + "evalId": "e1", + "checkThresholds": [{"checkId": "code-reading", "minimumPassRate": 0.98}], + "expectedUpdatedAt": None, + }, + ) + + def test_invalid_threshold_fails_before_any_request(self) -> None: + result = _gates_cli("workflow", "deploy-gates", "eval", "update", "my-wf", "--threshold", "0.98") + self.assertEqual(result.exit_code, 1) + self.assertIn("CHECK_ID=RATE", result.output) + self.assertEqual(len(responses.calls), 0) + + def test_refused_deploy_points_at_the_checks(self) -> None: + refusal = { + "error": { + "code": "deploy_check_not_passed", + "category": "conflict", + "retryable": False, + "requestId": "r", + "message": "The version has not passed its deploy gates", + "details": {"gateType": "workflow_eval", "nextSteps": []}, + } + } + responses.add(responses.POST, f"{GATES}/versions/v1/deploy", json=refusal, status=409) + result = _gates_cli("workflow", "deploy", "my-wf", "v1") + self.assertEqual(result.exit_code, 1) + self.assertIn("roboflow workflow deploy-gates run my-wf v1 --wait", result.output) class TestWorkflowNoWorkspace(unittest.TestCase): diff --git a/tests/test_workflow_deploy_gates.py b/tests/test_workflow_deploy_gates.py new file mode 100644 index 00000000..6619ea05 --- /dev/null +++ b/tests/test_workflow_deploy_gates.py @@ -0,0 +1,185 @@ +import json +import unittest +import uuid +from unittest.mock import patch + +import responses + +from roboflow.adapters.rfapi import RoboflowError +from roboflow.adapters.workflowdeploygatesapi import WorkflowDeployGateError +from roboflow.adapters.workflowevalsapi import WorkflowEvalError +from roboflow.config import API_URL +from roboflow.core.workflow_deploy_gates import WorkflowDeployGates + +WS = "test-ws" +KEY = "test_key" +BASE = f"{API_URL}/workspaces/{WS}/workflows" +WF, V = "wf-1", "ver-1" +GATE = f"/{WF}/deploy-gates/workflow-eval" + +# (client method, args, kwargs, HTTP method, path relative to BASE, sends Idempotency-Key) +# One row per route in the server's publicDeployGatesRegistry.ts. +ROUTES = [ + ("openapi", (), {}, "GET", "/deploy-gates/openapi.json", False), + ("list", (), {}, "GET", "/deploy-gates", False), + ("list_for_workflow", (WF,), {}, "GET", f"/{WF}/deploy-gates", False), + ("save_version", (WF, {"specification": {}}), {}, "POST", f"/{WF}/versions", True), + ("get_version", (WF, V), {}, "GET", f"/{WF}/versions/{V}/deploy-gates", False), + ("run", (WF, V), {}, "POST", f"/{WF}/versions/{V}/deploy-gates/run", True), + ("deploy", (WF, V), {}, "POST", f"/{WF}/versions/{V}/deploy", False), + ("get_eval_gate", (WF,), {}, "GET", GATE, False), + ("set_eval_gate", (WF, "eval-1"), {}, "PUT", GATE, False), + ("update_eval_gate", (WF, {"c1": 0.9}), {}, "PATCH", GATE, False), + ("set_eval_gate_new_check_thresholds", (WF, {"c1": 0.9}), {}, "POST", f"{GATE}/new-check-thresholds", False), + ("remove_eval_gate", (WF,), {}, "DELETE", GATE, False), + ("get_eval_checks", (WF,), {}, "GET", f"{GATE}/checks", False), +] + +NOT_PASSED = { + "error": { + "code": "deploy_check_not_passed", + "category": "conflict", + "retryable": False, + "requestId": "req-1", + "message": "The version has not passed its deploy gates", + "details": { + "gateType": "workflow_eval", + "status": "failed", + "current": True, + "staleReason": None, + "nextSteps": [ + { + "action": "run_deploy_checks", + "method": "POST", + "path": f"/workspaces/{WS}/workflows/{WF}/versions/{V}/deploy-gates/run", + "description": "Start every deploy gate of the workflow on the version", + } + ], + }, + } +} + + +def _is_uuid4(value): + try: + return uuid.UUID(value).version == 4 + except (TypeError, ValueError): + return False + + +def _verdicts(*statuses, publishable=False): + gates = [{"type": "workflow_eval", "status": status, "reason": None} for status in statuses] + return {"versionId": V, "gates": gates, "publishable": publishable} + + +class TestWorkflowDeployGatesRoutes(unittest.TestCase): + def setUp(self): + self.client = WorkflowDeployGates(KEY, WS) + + def test_covers_every_public_route(self): + self.assertEqual(len({(row[3], row[4]) for row in ROUTES}), 13) + + @responses.activate + def test_each_method_calls_its_route(self): + for name, args, kwargs, method, path, idempotent in ROUTES: + with self.subTest(method=name): + responses.reset() + responses.add(method, f"{BASE}{path}", json={"ok": True}, status=200) + getattr(self.client, name)(*args, **kwargs) + request = responses.calls[0].request + self.assertEqual(request.method, method) + self.assertEqual(request.url.split("?")[0], f"{BASE}{path}") + self.assertEqual(request.headers["Authorization"], f"Bearer {KEY}") + self.assertEqual(_is_uuid4(request.headers.get("Idempotency-Key")), idempotent) + + +class TestWorkflowDeployGatesClient(unittest.TestCase): + def setUp(self): + self.client = WorkflowDeployGates(KEY, WS) + + @responses.activate + def test_save_version_wraps_a_bare_definition_and_keeps_the_key(self): + responses.add(responses.POST, f"{BASE}/{WF}/versions", json={"replayed": False}, status=201) + bare = {"version": "1.0", "inputs": [], "steps": [], "outputs": []} + key = str(uuid.uuid4()) + self.client.save_version(WF, bare, version_name="candidate", idempotency_key=key) + request = responses.calls[0].request + body = json.loads(request.body) + self.assertEqual(json.loads(body["config"]), {"specification": bare}) + self.assertEqual(body["versionName"], "candidate") + self.assertEqual(request.headers["Idempotency-Key"], key) + + @responses.activate + def test_set_eval_gate_sends_thresholds_and_the_stale_write_guard(self): + responses.add(responses.PUT, f"{BASE}{GATE}", json={}, status=200) + cases = [ + ({}, {"evalId": "e1"}), + ({"expected_updated_at": None}, {"evalId": "e1", "expectedUpdatedAt": None}), + ( + {"check_thresholds": {"code": 0.98}, "expected_updated_at": "2026-11-04T10:00:00.000Z"}, + { + "evalId": "e1", + "checkThresholds": [{"checkId": "code", "minimumPassRate": 0.98}], + "expectedUpdatedAt": "2026-11-04T10:00:00.000Z", + }, + ), + ( + {"check_thresholds": [{"checkId": "code", "minimumPassRate": 0.5}]}, + {"evalId": "e1", "checkThresholds": [{"checkId": "code", "minimumPassRate": 0.5}]}, + ), + ] + for kwargs, expected in cases: + with self.subTest(kwargs=kwargs): + responses.calls.reset() + self.client.set_eval_gate(WF, "e1", **kwargs) + self.assertEqual(json.loads(responses.calls[0].request.body), expected) + + @responses.activate + def test_get_eval_checks_filters_by_version(self): + responses.add(responses.GET, f"{BASE}{GATE}/checks", json={}, status=200) + self.client.get_eval_checks(WF, version_id=V) + self.assertEqual(responses.calls[0].request.params, {"versionId": V}) + + @responses.activate + @patch("roboflow.core.workflow_deploy_gates.time.sleep") + def test_wait_for_version_polls_until_no_gate_is_pending(self, mock_sleep): + url = f"{BASE}/{WF}/versions/{V}/deploy-gates" + responses.add(responses.GET, url, json=_verdicts("pending"), status=200) + responses.add(responses.GET, url, json=_verdicts("pending"), status=200) + responses.add(responses.GET, url, json=_verdicts("failed"), status=200) + polled = [] + result = self.client.wait_for_version(WF, V, interval=0, on_poll=polled.append) + self.assertEqual(result["gates"][0]["status"], "failed") + self.assertEqual(len(polled), 3) + self.assertEqual(mock_sleep.call_count, 2) + + @responses.activate + def test_wait_for_version_returns_at_once_when_nothing_is_pending(self): + responses.add( + responses.GET, f"{BASE}/{WF}/versions/{V}/deploy-gates", json=_verdicts("not_started"), status=200 + ) + self.assertFalse(self.client.wait_for_version(WF, V)["publishable"]) + self.assertEqual(len(responses.calls), 1) + + @responses.activate + def test_wait_for_version_times_out(self): + responses.add(responses.GET, f"{BASE}/{WF}/versions/{V}/deploy-gates", json=_verdicts("pending"), status=200) + with self.assertRaises(TimeoutError): + self.client.wait_for_version(WF, V, timeout=0) + + @responses.activate + def test_refused_deploy_raises_the_typed_error(self): + responses.add(responses.POST, f"{BASE}/{WF}/versions/{V}/deploy", json=NOT_PASSED, status=409) + with self.assertRaises(WorkflowDeployGateError) as ctx: + self.client.deploy(WF, V) + error = ctx.exception + self.assertIsInstance(error, WorkflowEvalError) + self.assertIsInstance(error, RoboflowError) + self.assertEqual(error.status_code, 409) + self.assertEqual(error.code, "deploy_check_not_passed") + self.assertEqual(error.gate_types, ["workflow_eval"]) + self.assertEqual(error.next_steps[0]["action"], "run_deploy_checks") + + +if __name__ == "__main__": + unittest.main() From 5c9f876bd5619520301b5850946f30d01e186733 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o?= Date: Wed, 4 Nov 2026 23:24:33 -0500 Subject: [PATCH 2/4] docs: link the deploy gates changelog entries to #579 --- CHANGELOG.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 36699acc..24d8109c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,11 +13,13 @@ All notable changes to this project will be documented in this file. Workflow Eval gate is managed with `get_eval_gate`, `set_eval_gate`, `update_eval_gate`, `set_eval_gate_new_check_thresholds`, `remove_eval_gate` and `get_eval_checks`. Errors raise `WorkflowDeployGateError`, whose - `next_steps` lists what publishes the workflow. + `next_steps` lists what publishes the workflow + ([#579](https://github.com/roboflow/roboflow-python/pull/579)). - `roboflow workflow version save`, `roboflow workflow deploy-gates list|run|status`, `roboflow workflow deploy-gates eval get|set|update|new-check-thresholds|remove|checks` and `roboflow workflow deploy`, which replaces the hidden stub. `run --wait` and - `status --wait` exit with code 1 when the version is not publishable. + `status --wait` exit with code 1 when the version is not publishable + ([#579](https://github.com/roboflow/roboflow-python/pull/579)). - `Project.autolabel(..., keypoint_confidence_threshold=0.75)` and `roboflow autolabel start --keypoint-confidence 0.75` set the lowest confidence a predicted keypoint needs to be saved as visible when a keypoint-detection @@ -40,7 +42,7 @@ All notable changes to this project will be documented in this file. `code`, `gate_types` and `next_steps`, instead of a `RoboflowError` holding the raw JSON. The server saves nothing in that case: save a version, run its deploy checks and deploy it. Other `update_workflow` and `get_workflow` errors - now carry `status_code`. + now carry `status_code` ([#579](https://github.com/roboflow/roboflow-python/pull/579)). ### Deprecated From 37eaa2db89bd270783f84bf14704f8db6ca51ded Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o?= Date: Wed, 4 Nov 2026 23:27:22 -0500 Subject: [PATCH 3/4] style: move the UNSET note out of a docstring --- roboflow/core/workflow_deploy_gates.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/roboflow/core/workflow_deploy_gates.py b/roboflow/core/workflow_deploy_gates.py index 051fac48..f3eebe56 100644 --- a/roboflow/core/workflow_deploy_gates.py +++ b/roboflow/core/workflow_deploy_gates.py @@ -56,8 +56,8 @@ def __repr__(self) -> str: return "UNSET" +# Default of ``expected_updated_at``: leave the stale-write check out. UNSET: Any = _Unset() -"""Default of ``expected_updated_at``: leave the stale-write check out.""" def _thresholds(check_thresholds: ThresholdsArg) -> List[Dict[str, Any]]: From a7cd94ddff401adfcf587961a08f00e8f84a446d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o?= Date: Thu, 5 Nov 2026 17:12:53 -0500 Subject: [PATCH 4/4] feat(workflows): support the manual approval deploy gate roboflow/roboflow#18948 added a manual_approval gate. The client and CLI can now set, read and remove its rules and read version approvals; approvers still decide in the app. --- CHANGELOG.md | 10 +- CLI-COMMANDS.md | 8 +- docs/core/workspace.md | 15 ++ roboflow/adapters/workflowdeploygatesapi.py | 6 +- roboflow/cli/handlers/workflow.py | 224 ++++++++++++++++++-- roboflow/core/workflow_deploy_gates.py | 102 ++++++++- tests/cli/test_workflow_handler.py | 65 ++++-- tests/test_workflow_deploy_gates.py | 28 ++- 8 files changed, 412 insertions(+), 46 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 24d8109c..21a3185b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,11 +12,15 @@ All notable changes to this project will be documented in this file. verdicts, and `deploy` publishes a version once it passed every gate. The Workflow Eval gate is managed with `get_eval_gate`, `set_eval_gate`, `update_eval_gate`, `set_eval_gate_new_check_thresholds`, `remove_eval_gate` - and `get_eval_checks`. Errors raise `WorkflowDeployGateError`, whose - `next_steps` lists what publishes the workflow + and `get_eval_checks`. The manual approval gate is managed with + `get_approval_gate`, `set_approval_gate` and `remove_approval_gate`, and + `list_approvals` and `get_version_approval` read who approved a version. + Approvers decide in the Roboflow app. Errors raise `WorkflowDeployGateError`, + whose `next_steps` lists what publishes the workflow ([#579](https://github.com/roboflow/roboflow-python/pull/579)). - `roboflow workflow version save`, `roboflow workflow deploy-gates list|run|status`, - `roboflow workflow deploy-gates eval get|set|update|new-check-thresholds|remove|checks` + `roboflow workflow deploy-gates eval get|set|update|new-check-thresholds|remove|checks`, + `roboflow workflow deploy-gates approval get|set|remove|list|status` and `roboflow workflow deploy`, which replaces the hidden stub. `run --wait` and `status --wait` exit with code 1 when the version is not publishable ([#579](https://github.com/roboflow/roboflow-python/pull/579)). diff --git a/CLI-COMMANDS.md b/CLI-COMMANDS.md index 98a3c158..23e0b0c8 100644 --- a/CLI-COMMANDS.md +++ b/CLI-COMMANDS.md @@ -407,10 +407,16 @@ roboflow workflow deploy my-workflow roboflow workflow deploy-gates eval set my-workflow --eval --threshold code-reading=0.98 roboflow workflow deploy-gates eval get my-workflow roboflow workflow deploy-gates list + +# Gate a workflow on manual approval: one approval from anyone named approves a version. +roboflow workflow deploy-gates approval set my-workflow --user --role admin +roboflow workflow deploy-gates approval status my-workflow ``` On an Auto-Publish workflow a version that passes publishes itself, so `deploy` is only needed on -a Protected one. +a Protected one. Approvers approve or reject in the Roboflow app; the CLI only reads approvals, and +`run --wait` also waits for one. Setting or removing the approval rules needs a person, so an API +key is refused there. ### Fork a Universe project (async) diff --git a/docs/core/workspace.md b/docs/core/workspace.md index ac0e2629..f2bdb5e2 100644 --- a/docs/core/workspace.md +++ b/docs/core/workspace.md @@ -59,3 +59,18 @@ gates.set_eval_gate(workflow_id, "EVAL_ID", check_thresholds={"code-reading": 0. ``` Checks left out need every applicable Case to pass. + +A manual approval gate waits for one approval from a member or role its rules name. Approvers +decide in the Roboflow app, so `run` starts nothing for it and `wait_for_version` waits for the +approval: + +```python +gates.set_approval_gate(workflow_id, user_ids=["USER_ID"], roles=["admin"]) +approval = gates.get_version_approval(workflow_id, version_id) +print(approval["status"], approval["missing"]) +``` + +Approvers need the Manage Workflows permission, and the version's author cannot approve it unless +`allow_self_approval=True`. Changing the rules makes earlier decisions stop counting. Only a person +changes the rules: `set_approval_gate` and `remove_approval_gate` refuse a plain API key with +`deploy_approval_rules_user_required`. diff --git a/roboflow/adapters/workflowdeploygatesapi.py b/roboflow/adapters/workflowdeploygatesapi.py index 8714c8d3..f58b8d77 100644 --- a/roboflow/adapters/workflowdeploygatesapi.py +++ b/roboflow/adapters/workflowdeploygatesapi.py @@ -36,6 +36,10 @@ "deploy_gate_invalid_eval", "deploy_gate_feature_unavailable", "deploy_gate_permission_required", + "deploy_gate_invalid_rules", + "deploy_approval_rules_user_required", + "deploy_approval_not_allowed", + "deploy_approval_invalid_reviewers", } ) @@ -58,7 +62,7 @@ def next_steps(self) -> List[Dict[str, Any]]: @property def gate_types(self) -> List[str]: - """The gate types behind the refusal, e.g. ``["workflow_eval"]``.""" + """The gate types behind the refusal, e.g. ``["workflow_eval", "manual_approval"]``.""" gate_types = self.details.get("gateTypes") if isinstance(gate_types, list): return gate_types diff --git a/roboflow/cli/handlers/workflow.py b/roboflow/cli/handlers/workflow.py index d4df6050..d8644eb5 100644 --- a/roboflow/cli/handlers/workflow.py +++ b/roboflow/cli/handlers/workflow.py @@ -31,6 +31,13 @@ _eval_gate_app = typer.Typer(cls=SortedGroup, help="Gate a workflow on a Workflow Eval", no_args_is_help=True) _deploy_gates_app.add_typer(_eval_gate_app, name="eval") +_approval_gate_app = typer.Typer( + cls=SortedGroup, + help="Gate a workflow on manual approval (approvers decide in the Roboflow app)", + no_args_is_help=True, +) +_deploy_gates_app.add_typer(_approval_gate_app, name="approval") + WorkflowArg = Annotated[ str, typer.Argument(help="Workflow URL or ID (resolved with a GET, which needs the workflow:read scope)") ] @@ -162,9 +169,12 @@ def deploy_workflow(ctx: typer.Context, workflow_url: WorkflowArg, workflow_vers def list_deploy_gates( ctx: typer.Context, workflow_url: Annotated[ - Optional[str], typer.Argument(help="Workflow URL or ID; omit to list the whole workspace's gates") + Optional[str], typer.Argument(help="Workflow URL or ID; omit to list the workspace's Eval gates") + ] = None, + gate_type: Annotated[ + Optional[str], + typer.Option("--type", help="Only gates of this type (the workspace list has workflow_eval only)"), ] = None, - gate_type: Annotated[Optional[str], typer.Option("--type", help="Only gates of this type")] = None, eval_id: Annotated[Optional[str], typer.Option("--eval", help="Only gates backed by this Eval")] = None, ) -> None: """List deploy gates, of one workflow or of the whole workspace.""" @@ -182,7 +192,7 @@ def run_deploy_gates( interval: IntervalOpt = 5, idempotency_key: IdempotencyOpt = None, ) -> None: - """Start every deploy gate on a version (an Eval gate runs the Eval, which may consume credits).""" + """Start every deploy gate on a version. An Eval gate runs the Eval (may consume credits); approval waits.""" args = ctx_to_args( ctx, workflow_url=workflow_url, @@ -301,6 +311,88 @@ def list_eval_gate_checks( _gate_call(args, lambda gates, wf: gates.get_eval_checks(wf, version_id=version_id), _eval_checks_text) +@_approval_gate_app.command("get") +def get_approval_gate(ctx: typer.Context, workflow_url: WorkflowArg) -> None: + """Show who may approve the workflow's versions.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + _gate_call(args, lambda gates, wf: gates.get_approval_gate(wf), _approval_gate_text) + + +@_approval_gate_app.command("set") +def set_approval_gate( + ctx: typer.Context, + workflow_url: WorkflowArg, + user: Annotated[ + Optional[List[str]], typer.Option("--user", help="Member user ID who may approve (repeatable)") + ] = None, + role: Annotated[ + Optional[List[str]], typer.Option("--role", help="Role ID whose members may approve (repeatable)") + ] = None, + allow_self_approval: Annotated[ + Optional[bool], + typer.Option( + "--self-approval/--no-self-approval", + help="Whether a version's author may approve it (server default: no)", + ), + ] = None, + expected_revision: Annotated[ + Optional[int], + typer.Option("--expected-revision", help="The gate's revision as you read it; refuses the write if it changed"), + ] = None, +) -> None: + """Set who may approve the workflow's versions; one approval is enough. A plain API key is refused.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + if not user and not role: + from roboflow.cli._output import output_error + + output_error(args, "Name at least one approver.", hint="Pass --user USER_ID or --role ROLE_ID (repeatable).") + return + _gate_call( + args, + lambda gates, wf: gates.set_approval_gate( + wf, + user_ids=user or (), + roles=role or (), + allow_self_approval=allow_self_approval, + expected_revision=expected_revision, + ), + _approval_gate_text, + ) + + +@_approval_gate_app.command("remove") +def remove_approval_gate( + ctx: typer.Context, + workflow_url: WorkflowArg, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt.")] = False, +) -> None: + """Remove the manual approval gate, so versions publish without approval.""" + from roboflow.cli._output import confirm_destructive + + args = ctx_to_args(ctx, workflow_url=workflow_url, yes=yes) + if not confirm_destructive(args, f"Remove the manual approval deploy gate from '{workflow_url}'?"): + return + _gate_call( + args, + lambda gates, wf: gates.remove_approval_gate(wf), + lambda _status: f"Removed the manual approval deploy gate from {workflow_url}.", + ) + + +@_approval_gate_app.command("list") +def list_approvals(ctx: typer.Context, workflow_url: WorkflowArg) -> None: + """Show the approval status of every version someone decided on or was asked to review.""" + args = ctx_to_args(ctx, workflow_url=workflow_url) + _gate_call(args, lambda gates, wf: gates.list_approvals(wf), _approvals_text) + + +@_approval_gate_app.command("status") +def version_approval_status(ctx: typer.Context, workflow_url: WorkflowArg, workflow_version: VersionArg) -> None: + """Show who approved or rejected a version, and who may still approve it.""" + args = ctx_to_args(ctx, workflow_url=workflow_url, version_id=workflow_version) + _gate_call(args, lambda gates, wf: gates.get_version_approval(wf, workflow_version), _version_approval_text) + + @workflow_app.command("delete") def delete_workflow( ctx: typer.Context, @@ -753,25 +845,51 @@ def _resolve_workflow_id(args, api_key: str, workspace_url: str) -> Optional[str return str(workflow_id) +def _gate_steps(workflow: str, version: str, gate_types: List[str]) -> List[str]: + """The CLI steps that pass the refusing gates; no gate types means the Eval gate's.""" + run = f"roboflow workflow deploy-gates run {workflow} {version} --wait" + approved = ( + "an approver approves it in the Roboflow app " + f"(roboflow workflow deploy-gates approval status {workflow} {version} shows who may)" + ) + if "manual_approval" not in gate_types: + return [run] + if "workflow_eval" not in gate_types: + return [f"wait until {approved}"] + return [f"{run}, which also waits until {approved}"] + + def _gate_hint(args, exc) -> Optional[str]: # noqa: ANN001 workflow = getattr(args, "workflow_url", None) or "" version = getattr(args, "version_id", None) or "" definition = getattr(args, "definition", None) or "" + gate_types = getattr(exc, "gate_types", []) + publish_steps = [ + f"roboflow workflow version save {workflow} --definition {definition}", + *_gate_steps(workflow, "", gate_types), + f"roboflow workflow deploy {workflow} ", + ] + pass_steps = _gate_steps(workflow, version, gate_types) hints = { "deploy_check_required": ( "This workflow has deploy gates, so it publishes only a saved version that passed them. Run: " - f"roboflow workflow version save {workflow} --definition {definition}, then " - f"roboflow workflow deploy-gates run {workflow} --wait, then " - f"roboflow workflow deploy {workflow} ." - ), - "deploy_check_not_passed": ( - f"Run the version's deploy checks first: roboflow workflow deploy-gates run {workflow} {version} --wait." + + ", then ".join(publish_steps) + + "." ), + "deploy_check_not_passed": "Pass the version's deploy gates first: " + ", then ".join(pass_steps) + ".", "deploy_gate_feature_unavailable": "Deploy gates need a plan with workflow version history.", "deploy_gate_permission_required": "Changing deploy gates needs the Manage Workflow Deploy Gates permission.", "deploy_gate_invalid_eval": "The Eval cannot gate this workflow; see error.details.reason.", "deploy_check_local_inference_not_allowed": "Deploy checks run on the server. Set the Eval to run there.", "deploy_check_unavailable": "Deploy checks cannot start right now. Try again later.", + "deploy_gate_invalid_rules": ( + "The approval rules cannot be set; see error.details.reason. Approvers must be workspace members " + "whose role grants Manage Workflows." + ), + "deploy_approval_rules_user_required": ( + "Only a person can change who approves, and an API key is not one. Set the approval rules in the " + "Roboflow app, or with a credential that acts for a user." + ), } if exc.code in hints: return hints[exc.code] @@ -962,16 +1080,90 @@ def _workflow_gates_text(data: Dict[str, Any]) -> str: rows = [] for item in data.get("items") or []: status = item.get("status") or {} - evaluation = status.get("eval") or {} - rows.append( + if item.get("type") == "manual_approval": + details = "approvers: " + ", ".join(row["approver"] for row in _approver_rows(status)) + else: + details = "eval: " + ( + (status.get("eval") or {}).get("name") or (status.get("gate") or {}).get("evalId", "") + ) + rows.append({"type": item.get("type", ""), "details": details}) + if not rows: + return "The workflow has no deploy gates." + return format_table(rows, columns=["type", "details"], headers=["GATE", "DETAILS"]) + + +def _approver_rows(status: Dict[str, Any]) -> List[Dict[str, str]]: + """One row per approval rule, naming the member or role as the app shows it.""" + users = {u.get("uid"): u for u in status.get("users") or []} + roles = {r.get("id"): r for r in status.get("roles") or []} + rows = [] + for rule in (status.get("gate") or {}).get("rules") or []: + if rule.get("kind") == "user": + uid = rule.get("userId", "") + member = users.get(uid) or {} + rows.append({"kind": "user", "approver": member.get("email") or member.get("name") or uid, "id": uid}) + else: + role_id = rule.get("role", "") + rows.append({"kind": "role", "approver": (roles.get(role_id) or {}).get("name") or role_id, "id": role_id}) + return rows + + +def _approval_gate_text(status: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + gate = status.get("gate") + if not gate: + return "No manual approval gate." + self_approval = "allowed" if gate.get("allowSelfApproval") else "not allowed" + lines = [ + f"Revision: {gate.get('revision', '')} (updated {gate.get('updatedAt', '')})", + f"Self-approval: {self_approval}", + "One approval from anyone below approves a version; one rejection rejects it.", + "", + format_table(_approver_rows(status), columns=["kind", "approver", "id"], headers=["RULE", "APPROVER", "ID"]), + ] + return "\n".join(lines) + + +def _version_approval_text(approval: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + lines = [f"Version {approval.get('versionId', '')}: {approval.get('status', '')}"] + if approval.get("reason"): + lines.append(str(approval["reason"])) + decisions = approval.get("decisions") or [] + if decisions: + rows = [ { - "type": item.get("type", ""), - "eval": evaluation.get("name") or (status.get("gate") or {}).get("evalId", ""), + "approver": (d.get("approver") or {}).get("email") or (d.get("approver") or {}).get("uid", ""), + "decision": d.get("decision", ""), + "counted": "yes" if d.get("counted") else f"no ({d.get('staleReason') or 'not eligible'})", + "decided": d.get("decidedAt", ""), + "comment": d.get("comment") or "", } - ) + for d in decisions + ] + lines += [ + "", + format_table( + rows, + columns=["approver", "decision", "counted", "decided", "comment"], + headers=["APPROVER", "DECISION", "COUNTED", "DECIDED", "COMMENT"], + ), + ] + return "\n".join(lines) + + +def _approvals_text(data: Dict[str, Any]) -> str: + from roboflow.cli._table import format_table + + rows = [ + {"version": a.get("versionId", ""), "status": a.get("status", ""), "reason": a.get("reason") or ""} + for a in data.get("items") or [] + ] if not rows: - return "The workflow has no deploy gates." - return format_table(rows, columns=["type", "eval"], headers=["GATE", "EVAL"]) + return "No version has an approval decision or review request." + return format_table(rows, columns=["version", "status", "reason"], headers=["VERSION", "STATUS", "REASON"]) def _save_workflow_version(args) -> None: # noqa: ANN001 diff --git a/roboflow/core/workflow_deploy_gates.py b/roboflow/core/workflow_deploy_gates.py index f3eebe56..6a9acce4 100644 --- a/roboflow/core/workflow_deploy_gates.py +++ b/roboflow/core/workflow_deploy_gates.py @@ -1,10 +1,15 @@ """Workflow deploy gates: wraps the public ``/workspaces/{workspace}/workflows`` gate API. A deploy gate is a condition a saved Workflow version must meet before it -publishes. Today the only type is ``workflow_eval``: a fresh Run of the version -on one Workflow Eval must reach every check's minimum pass rate. While a -workflow has a gate, ``updateWorkflow`` no longer publishes its config; a -version publishes only through :meth:`WorkflowDeployGates.deploy`, once it +publishes. A workflow has at most one gate of each type: + +* ``workflow_eval``: a fresh Run of the version on one Workflow Eval must reach + every check's minimum pass rate. +* ``manual_approval``: one of the members or roles the rules name approves the + version in the Roboflow app. This API reads approvals but cannot decide. + +While a workflow has a gate, ``updateWorkflow`` no longer publishes its config; +a version publishes only through :meth:`WorkflowDeployGates.deploy`, once it passed every gate. Get a client from :meth:`roboflow.core.workspace.Workspace.workflow_deploy_gates`:: @@ -36,7 +41,9 @@ also needs ``workflow-evals:run`` and ``model:infer``, and :meth:`set_eval_gate_new_check_thresholds` also ``workflow-evals:write``. A check's Eval Run results come back only with ``workflow-evals:read``. -Folder-scoped keys are refused. +Folder-scoped keys are refused. Only a person changes who approves: +:meth:`set_approval_gate` and :meth:`remove_approval_gate` refuse a plain API +key with ``403 deploy_approval_rules_user_required``. """ from __future__ import annotations @@ -91,6 +98,10 @@ def _version_path(cls, workflow_id: str, version_id: str) -> str: def _eval_gate_path(cls, workflow_id: str) -> str: return f"{cls._workflow_path(workflow_id)}/deploy-gates/workflow-eval" + @classmethod + def _approval_gate_path(cls, workflow_id: str) -> str: + return f"{cls._workflow_path(workflow_id)}/deploy-gates/manual-approval" + # -- discovery ---------------------------------------------------------- def openapi(self) -> Dict[str, Any]: @@ -98,7 +109,11 @@ def openapi(self) -> Dict[str, Any]: return self._call("GET", "/deploy-gates/openapi.json") def list(self, *, gate_type: Optional[str] = None, eval_id: Optional[str] = None) -> Dict[str, Any]: - """List the workspace's deploy gates, optionally of one type or backed by one Eval.""" + """List the workspace's Eval deploy gates, optionally only those backed by one Eval. + + The server lists ``workflow_eval`` gates only, so ``gate_type`` accepts + that value alone. + """ return self._call("GET", "/deploy-gates", params={"type": gate_type, "evalId": eval_id}) def list_for_workflow(self, workflow_id: str) -> Dict[str, Any]: @@ -148,7 +163,8 @@ def run(self, workflow_id: str, version_id: str, *, idempotency_key: Optional[st even a failed one, and never starts another Run. A new key reuses a pending or passed check and starts a new one on a failed, errored, cancelled or outdated check. A Workflow Eval gate runs the Eval on the - server, which may consume credits. + server, which may consume credits. A manual approval gate starts + nothing: the version waits for an approver. """ return self._call( "POST", @@ -169,9 +185,10 @@ def wait_for_version( """Poll :meth:`get_version` until no gate is ``pending`` and return the verdicts. Check ``publishable`` on the result: a gate can end ``failed``, or - ``not_started`` when nothing was started. Stopping the wait never - cancels a check. Raises ``TimeoutError`` when ``timeout`` seconds elapse - first. + ``not_started`` when nothing was started. A manual approval gate stays + ``pending`` until someone approves or rejects the version in the app, so + this also waits for that. Stopping the wait never cancels a check. + Raises ``TimeoutError`` when ``timeout`` seconds elapse first. """ deadline = time.monotonic() + timeout while True: @@ -261,3 +278,68 @@ def get_eval_checks(self, workflow_id: str, *, version_id: Optional[str] = None) With ``version_id`` the server answers 404 until a check was started. """ return self._call("GET", f"{self._eval_gate_path(workflow_id)}/checks", params={"versionId": version_id}) + + # -- manual_approval gate ----------------------------------------------- + + def get_approval_gate(self, workflow_id: str) -> Dict[str, Any]: + """Return the approval rules with the members and roles they name: ``{gate, users, roles}``. + + ``gate`` is ``{revision, rules, allowSelfApproval, updatedAt, updatedBy}``, + or null without a manual approval gate. + """ + return self._call("GET", self._approval_gate_path(workflow_id)) + + def set_approval_gate( + self, + workflow_id: str, + *, + user_ids: Iterable[str] = (), + roles: Iterable[str] = (), + allow_self_approval: Optional[bool] = None, + expected_revision: Optional[int] = None, + ) -> Dict[str, Any]: + """Gate the workflow on approvals and set who may approve its versions. + + One approval from any member in ``user_ids``, or from any member whose + current role id is in ``roles``, approves a version; one rejection + rejects it. Between them, 1 to 10 rules. An approver needs the Manage + Workflows permission. The version's author cannot approve it unless + ``allow_self_approval`` is true (the server default is false). + + ``expected_revision`` is the gate's ``revision`` as you read it: the + write is refused with ``409 CONFLICT`` if the gate changed since. Leave + it out to replace whatever is stored. Changing the rules makes earlier + decisions stop counting. Refused for a plain API key + (``403 deploy_approval_rules_user_required``). + """ + rules = [{"kind": "user", "userId": uid} for uid in user_ids] + rules += [{"kind": "role", "role": role} for role in roles] + body: Dict[str, Any] = {"rules": rules} + if allow_self_approval is not None: + body["allowSelfApproval"] = allow_self_approval + if expected_revision is not None: + body["expectedRevision"] = expected_revision + return self._call("PUT", self._approval_gate_path(workflow_id), body=body) + + def remove_approval_gate(self, workflow_id: str) -> Dict[str, Any]: + """Remove the manual approval gate and return the gate status after the removal. + + Refused for a plain API key (``403 deploy_approval_rules_user_required``). + """ + return self._call("DELETE", self._approval_gate_path(workflow_id)) + + def list_approvals(self, workflow_id: str) -> Dict[str, Any]: + """Return ``{"items": [...]}``, one approval status per version with a decision or review request. + + Newest first; empty without a manual approval gate. + """ + return self._call("GET", f"{self._approval_gate_path(workflow_id)}/approvals") + + def get_version_approval(self, workflow_id: str, version_id: str) -> Dict[str, Any]: + """Return the version's approval status: who approved or rejected it, who may still approve. + + ``status`` is ``approved``, ``pending`` or ``rejected``; ``missing`` + names who is still needed; ``decisions`` (newest first) say whether each + one still counts. 404 on a workflow without a manual approval gate. + """ + return self._call("GET", f"{self._version_path(workflow_id, version_id)}/deploy-gates/manual-approval") diff --git a/tests/cli/test_workflow_handler.py b/tests/cli/test_workflow_handler.py index af5f4487..260230f8 100644 --- a/tests/cli/test_workflow_handler.py +++ b/tests/cli/test_workflow_handler.py @@ -87,6 +87,7 @@ def test_deploy_gate_commands_exist(self) -> None: ["deploy-gates", "eval", c] for c in ("get", "set", "update", "new-check-thresholds", "remove", "checks") ), + *(["deploy-gates", "approval", c] for c in ("get", "set", "remove", "list", "status")), ): with self.subTest(command=" ".join(argv)): result = runner.invoke(app, ["workflow", *argv, "--help"]) @@ -490,21 +491,57 @@ def test_invalid_threshold_fails_before_any_request(self) -> None: self.assertIn("CHECK_ID=RATE", result.output) self.assertEqual(len(responses.calls), 0) - def test_refused_deploy_points_at_the_checks(self) -> None: - refusal = { - "error": { - "code": "deploy_check_not_passed", - "category": "conflict", - "retryable": False, - "requestId": "r", - "message": "The version has not passed its deploy gates", - "details": {"gateType": "workflow_eval", "nextSteps": []}, - } - } - responses.add(responses.POST, f"{GATES}/versions/v1/deploy", json=refusal, status=409) - result = _gates_cli("workflow", "deploy", "my-wf", "v1") + def test_refused_deploy_points_at_what_passes_the_refusing_gate(self) -> None: + run = "roboflow workflow deploy-gates run my-wf v1 --wait" + approval = "roboflow workflow deploy-gates approval status my-wf v1" + for gate_type, shown, hidden in (("workflow_eval", run, approval), ("manual_approval", approval, run)): + with self.subTest(gate_type=gate_type): + responses.reset() + responses.add(responses.GET, f"{API_URL}/test-ws/workflows/my-wf", json=WORKFLOW, status=200) + refusal = { + "error": { + "code": "deploy_check_not_passed", + "category": "conflict", + "retryable": False, + "requestId": "r", + "message": "The version has not passed its deploy gates", + "details": {"gateType": gate_type, "nextSteps": []}, + } + } + responses.add(responses.POST, f"{GATES}/versions/v1/deploy", json=refusal, status=409) + result = _gates_cli("workflow", "deploy", "my-wf", "v1") + self.assertEqual(result.exit_code, 1) + self.assertIn(shown, result.output) + self.assertNotIn(hidden, result.output) + + def test_approval_set_sends_users_then_roles(self) -> None: + responses.add(responses.PUT, f"{GATES}/deploy-gates/manual-approval", json={"gate": None}, status=200) + result = _gates_cli( + "workflow", + "deploy-gates", + "approval", + "set", + "my-wf", + "--role", + "admin", + "--user", + "u1", + "--no-self-approval", + ) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + json.loads(responses.calls[1].request.body), + { + "rules": [{"kind": "user", "userId": "u1"}, {"kind": "role", "role": "admin"}], + "allowSelfApproval": False, + }, + ) + + def test_approval_set_without_approvers_fails_before_any_request(self) -> None: + result = _gates_cli("workflow", "deploy-gates", "approval", "set", "my-wf") self.assertEqual(result.exit_code, 1) - self.assertIn("roboflow workflow deploy-gates run my-wf v1 --wait", result.output) + self.assertIn("--user", result.output) + self.assertEqual(len(responses.calls), 0) class TestWorkflowNoWorkspace(unittest.TestCase): diff --git a/tests/test_workflow_deploy_gates.py b/tests/test_workflow_deploy_gates.py index 6619ea05..dd727510 100644 --- a/tests/test_workflow_deploy_gates.py +++ b/tests/test_workflow_deploy_gates.py @@ -16,6 +16,7 @@ BASE = f"{API_URL}/workspaces/{WS}/workflows" WF, V = "wf-1", "ver-1" GATE = f"/{WF}/deploy-gates/workflow-eval" +APPROVAL = f"/{WF}/deploy-gates/manual-approval" # (client method, args, kwargs, HTTP method, path relative to BASE, sends Idempotency-Key) # One row per route in the server's publicDeployGatesRegistry.ts. @@ -33,6 +34,11 @@ ("set_eval_gate_new_check_thresholds", (WF, {"c1": 0.9}), {}, "POST", f"{GATE}/new-check-thresholds", False), ("remove_eval_gate", (WF,), {}, "DELETE", GATE, False), ("get_eval_checks", (WF,), {}, "GET", f"{GATE}/checks", False), + ("get_approval_gate", (WF,), {}, "GET", APPROVAL, False), + ("set_approval_gate", (WF,), {"user_ids": ["u1"]}, "PUT", APPROVAL, False), + ("remove_approval_gate", (WF,), {}, "DELETE", APPROVAL, False), + ("list_approvals", (WF,), {}, "GET", f"{APPROVAL}/approvals", False), + ("get_version_approval", (WF, V), {}, "GET", f"/{WF}/versions/{V}/deploy-gates/manual-approval", False), ] NOT_PASSED = { @@ -77,7 +83,7 @@ def setUp(self): self.client = WorkflowDeployGates(KEY, WS) def test_covers_every_public_route(self): - self.assertEqual(len({(row[3], row[4]) for row in ROUTES}), 13) + self.assertEqual(len({(row[3], row[4]) for row in ROUTES}), 18) @responses.activate def test_each_method_calls_its_route(self): @@ -134,6 +140,26 @@ def test_set_eval_gate_sends_thresholds_and_the_stale_write_guard(self): self.client.set_eval_gate(WF, "e1", **kwargs) self.assertEqual(json.loads(responses.calls[0].request.body), expected) + @responses.activate + def test_set_approval_gate_sends_rules_and_only_the_options_given(self): + responses.add(responses.PUT, f"{BASE}{APPROVAL}", json={}, status=200) + cases = [ + ({"user_ids": ["u1"]}, {"rules": [{"kind": "user", "userId": "u1"}]}), + ( + {"user_ids": ["u1"], "roles": ["admin"], "allow_self_approval": False, "expected_revision": 3}, + { + "rules": [{"kind": "user", "userId": "u1"}, {"kind": "role", "role": "admin"}], + "allowSelfApproval": False, + "expectedRevision": 3, + }, + ), + ] + for kwargs, expected in cases: + with self.subTest(kwargs=kwargs): + responses.calls.reset() + self.client.set_approval_gate(WF, **kwargs) + self.assertEqual(json.loads(responses.calls[0].request.body), expected) + @responses.activate def test_get_eval_checks_filters_by_version(self): responses.add(responses.GET, f"{BASE}{GATE}/checks", json={}, status=200)