diff --git a/CHANGELOG.md b/CHANGELOG.md index ff78cbbf..2160e52f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,19 @@ All notable changes to this project will be documented in this file. +## Unreleased + +### Added + +- `roboflow api-key roll ` revokes an API key and creates its replacement + in one step, printing the new secret once. `--rotation-due-at` and `--owner` + (repeatable) set the replacement's rotation schedule; omitted, the old key's + schedule carries over. A protected key returns exit code 1 with a hint, like + `revoke`; a key that was already rolled or revoked is not found (exit code 3). +- `roboflow api-key create` and `update` accept `--rotation-due-at` and `--owner` + (repeatable) to set a key's rotation schedule; `update` also takes + `--clear-rotation-due-at` and `--clear-owners`. + ## 1.7.2 ### Fixed diff --git a/roboflow/adapters/rfapi.py b/roboflow/adapters/rfapi.py index 35177ab3..d7b69e0e 100644 --- a/roboflow/adapters/rfapi.py +++ b/roboflow/adapters/rfapi.py @@ -2473,6 +2473,19 @@ def __repr__(self) -> str: FULL_ACCESS = _FullAccess() +class _ClearValue: + """Sentinel for "send JSON ``null``" on fields where ``None`` means "leave unchanged". + + ``update_api_key(..., rotation_due_at=CLEAR)`` clears the key's rotation due date. + """ + + def __repr__(self) -> str: + return "CLEAR" + + +CLEAR = _ClearValue() + + def list_api_keys( api_key: str, workspace_url: str, @@ -2516,6 +2529,8 @@ def create_api_key( folder_ids: Optional[List[str]] = None, custom_metadata: Optional[Dict] = None, protected: bool = False, + rotation_due_at: Optional[str] = None, + owners: Optional[List[str]] = None, ) -> dict: """POST /{workspace}/api-keys — create a new API key. @@ -2524,8 +2539,11 @@ def create_api_key( own scopes, so a full-access credential creates a full-access key. Pass a list to scope the key (``role:`` presets are accepted), ``[]`` for a key with no abilities, or ``FULL_ACCESS`` to send an explicit ``null`` (unscoped/full - access). ``scopes``, ``folder_ids``, and ``custom_metadata`` require the - Advanced API Keys plan feature — the backend returns 403 if unavailable. + access). ``rotation_due_at`` (ISO date or date-time, UTC, in the future) and + ``owners`` (member ids reminded before it; ``[]`` means everyone who can roll + the key) set the rotation schedule. ``scopes``, ``folder_ids``, + ``custom_metadata``, ``rotation_due_at`` and ``owners`` require the Advanced + API Keys plan feature — the backend returns 403 if unavailable. """ body: Dict[str, Any] = {} if name is not None: @@ -2541,6 +2559,10 @@ def create_api_key( body["customMetadata"] = custom_metadata if protected: body["protected"] = True + if rotation_due_at is not None: + body["rotationDueAt"] = rotation_due_at + if owners is not None: + body["owners"] = owners response = http.post(f"{API_URL}/{workspace_url}/api-keys", params={"api_key": api_key}, json=body) if not response.ok: raise RoboflowError(response.text, status_code=response.status_code) @@ -2551,21 +2573,24 @@ def update_api_key(api_key: str, workspace_url: str, key_id: str, **fields: Any) """PATCH /{workspace}/api-keys/{keyId} — update an existing API key. Pass only the fields you want to change as keyword arguments: - ``name``, ``scopes``, ``custom_metadata``, ``protected``, ``disabled``. + ``name``, ``scopes``, ``custom_metadata``, ``protected``, ``disabled``, + ``rotation_due_at``, ``owners``. ``None`` values are omitted (left unchanged). To send explicit values, pass ``scopes=[]`` (no abilities), ``scopes=FULL_ACCESS`` (unscoped/full - access, serialized as ``null``), or ``custom_metadata={}`` (clear metadata). + access, serialized as ``null``), ``custom_metadata={}`` (clear metadata), + ``rotation_due_at=CLEAR`` (clear the due date, serialized as ``null``) or + ``owners=[]`` (remind everyone who can roll the key). The API cannot unprotect a key (``protected=False`` → 403). Disabling a protected key returns 409. """ encoded = quote(key_id, safe="") # Canonical wire field is camelCase `customMetadata` (consistent with `folderIds`); callers may # pass the Pythonic `custom_metadata` kwarg, which is normalized here. - wire_aliases = {"custom_metadata": "customMetadata"} + wire_aliases = {"custom_metadata": "customMetadata", "rotation_due_at": "rotationDueAt"} body: Dict[str, Any] = {} for k, v in fields.items(): wire_key = wire_aliases.get(k, k) - if v is FULL_ACCESS: + if v is FULL_ACCESS or v is CLEAR: body[wire_key] = None elif v is not None: body[wire_key] = v @@ -2592,3 +2617,36 @@ def revoke_api_key(api_key: str, workspace_url: str, key_id: str) -> dict: if not response.ok: raise RoboflowError(response.text, status_code=response.status_code) return response.json() + + +def roll_api_key( + api_key: str, + workspace_url: str, + key_id: str, + rotation_due_at: Optional[str] = None, + owners: Optional[List[str]] = None, +) -> dict: + """POST /{workspace}/api-keys/{keyId}/roll — revoke a key and mint its replacement. + + The replacement keeps the old key's name, scopes, folders, metadata, protection + and disabled state, and its secret ``key`` value is returned only here (shown + once). ``rotation_due_at`` (ISO date or date-time, UTC, in the future) and + ``owners`` set the replacement's rotation schedule; omitted, the old key's + schedule carries over. Both require the Advanced API Keys plan feature. + Rolling a protected or already rolled key returns 409; a credential that holds + less access than the target key gets 403. + """ + encoded = quote(key_id, safe="") + body: Dict[str, Any] = {} + if rotation_due_at is not None: + body["rotationDueAt"] = rotation_due_at + if owners is not None: + body["owners"] = owners + response = http.post( + f"{API_URL}/{workspace_url}/api-keys/{encoded}/roll", + params={"api_key": api_key}, + json=body or None, + ) + if not response.ok: + raise RoboflowError(response.text, status_code=response.status_code) + return response.json() diff --git a/roboflow/cli/handlers/api_key.py b/roboflow/cli/handlers/api_key.py index d4831492..484e055f 100644 --- a/roboflow/cli/handlers/api_key.py +++ b/roboflow/cli/handlers/api_key.py @@ -82,6 +82,17 @@ def create_key( bool, typer.Option("--protected", help="Mark key as protected (cannot be revoked/disabled via CLI)"), ] = False, + rotation_due_at: Annotated[ + Optional[str], + typer.Option( + "--rotation-due-at", + help="Rotation due date (ISO date, UTC, in the future). Owners get one reminder email before it.", + ), + ] = None, + owner: Annotated[ + Optional[List[str]], + typer.Option("--owner", help="Member ID reminded before the due date (repeatable). Omit to default to you."), + ] = None, ) -> None: """Create a new API key. @@ -93,6 +104,9 @@ def create_key( Scope selection is three-way: omit all scope flags to inherit the calling key's scopes, pass --scope (repeatable) to scope the key, --no-scopes for a key with no abilities, or --full-access for an unscoped/full-access key. + + --rotation-due-at sets the date the key should be rotated by; --owner + (repeatable) picks who gets the reminder email. Both require Advanced API Keys. """ args = ctx_to_args( ctx, @@ -103,6 +117,8 @@ def create_key( folder=folder, metadata=metadata, protected=protected, + rotation_due_at=rotation_due_at, + owner=owner, ) _create_key(args) @@ -144,13 +160,38 @@ def update_key( help="Clear all custom metadata (sends {}). Mutually exclusive with --metadata.", ), ] = False, + rotation_due_at: Annotated[ + Optional[str], + typer.Option("--rotation-due-at", help="New rotation due date (ISO date, UTC, in the future)."), + ] = None, + clear_rotation_due_at: Annotated[ + bool, + typer.Option( + "--clear-rotation-due-at", + help="Remove the rotation due date (sends null). Mutually exclusive with --rotation-due-at.", + ), + ] = False, + owner: Annotated[ + Optional[List[str]], + typer.Option("--owner", help="Member ID reminded before the due date (repeatable). Replaces the owners."), + ] = None, + clear_owners: Annotated[ + bool, + typer.Option( + "--clear-owners", + help="Remind everyone who can roll the key (sends []). Mutually exclusive with --owner.", + ), + ] = False, ) -> None: - """Update an API key's display name, scopes, or metadata. + """Update an API key's display name, scopes, metadata or rotation schedule. Scopes are three-way: --scope (repeatable) replaces scopes, --no-scopes replaces them with an empty list (no abilities), and --full-access makes the key unscoped/full access. Metadata: --metadata sets custom KEY=VALUE pairs, - --clear-metadata removes all custom metadata. + --clear-metadata removes all custom metadata. Rotation: --rotation-due-at + sets the date, --clear-rotation-due-at removes it; --owner (repeatable) + replaces the reminded members, --clear-owners reminds everyone who can roll + the key. Rotation fields require Advanced API Keys. """ args = ctx_to_args( ctx, @@ -161,6 +202,10 @@ def update_key( full_access=full_access, metadata=metadata, clear_metadata=clear_metadata, + rotation_due_at=rotation_due_at, + clear_rotation_due_at=clear_rotation_due_at, + owner=owner, + clear_owners=clear_owners, ) _update_key(args) @@ -210,6 +255,39 @@ def revoke_key( _revoke_key(args) +@api_key_app.command("roll") +def roll_key( + ctx: typer.Context, + key_id: Annotated[str, typer.Argument(help="Key ID (keyId) to roll")], + rotation_due_at: Annotated[ + Optional[str], + typer.Option( + "--rotation-due-at", + help="Rotation due date for the replacement (ISO date, UTC). Omit to carry the old key's date over.", + ), + ] = None, + owner: Annotated[ + Optional[List[str]], + typer.Option( + "--owner", help="Member ID reminded before the due date (repeatable). Omit to keep the old key's owners." + ), + ] = None, + yes: Annotated[bool, typer.Option("--yes", "-y", help="Skip confirmation prompt")] = False, +) -> None: + """Roll an API key: revoke it and create its replacement in one step. + + The replacement keeps the old key's name, scopes, folders, metadata and + protection. Everything using the old key stops working until it gets the + new value. The new secret is printed ONCE — save it immediately: + + roboflow --json api-key roll KEY_ID --yes | jq -r .key + + Protected keys cannot be rolled — unprotect them in the dashboard first. + """ + args = ctx_to_args(ctx, key_id=key_id, rotation_due_at=rotation_due_at, owner=owner, yes=yes) + _roll_key(args) + + # --------------------------------------------------------------------------- # Business logic # --------------------------------------------------------------------------- @@ -309,6 +387,49 @@ def _resolve_metadata(args): # noqa: ANN001 return _UNSET +def _resolve_rotation(args) -> dict: # noqa: ANN001 + """Resolve the rotation flags into ``update_api_key`` keyword arguments. + + ``--rotation-due-at`` sets the date and ``--clear-rotation-due-at`` sends + ``CLEAR`` (JSON null); ``--owner`` replaces the owners and ``--clear-owners`` + sends ``[]``. Flags not given are left out (unchanged). Exits 1 when a set and + a clear flag for the same field are both supplied. + """ + from roboflow.adapters.rfapi import CLEAR + from roboflow.cli._output import output_error + + rotation_due_at = getattr(args, "rotation_due_at", None) + clear_due = getattr(args, "clear_rotation_due_at", False) + owners = getattr(args, "owner", None) + clear_owners = getattr(args, "clear_owners", False) + + if rotation_due_at is not None and clear_due: + output_error( + args, + "Only one of --rotation-due-at or --clear-rotation-due-at may be used.", + hint="Use --rotation-due-at DATE to set the date, or --clear-rotation-due-at to remove it.", + exit_code=1, + ) + if owners and clear_owners: + output_error( + args, + "Only one of --owner or --clear-owners may be used.", + hint="Use --owner MEMBER_ID to pick the reminded members, or --clear-owners to remind everyone.", + exit_code=1, + ) + + fields: dict = {} + if clear_due: + fields["rotation_due_at"] = CLEAR + elif rotation_due_at is not None: + fields["rotation_due_at"] = rotation_due_at + if clear_owners: + fields["owners"] = [] + elif owners: + fields["owners"] = list(owners) + return fields + + def _list_keys(args) -> None: # noqa: ANN001 from roboflow.adapters import rfapi from roboflow.cli._output import output, output_api_error @@ -413,6 +534,7 @@ def _create_key(args) -> None: # noqa: ANN001 scopes = _resolve_scopes(args) folder_ids = getattr(args, "folder", None) or None metadata = _resolve_metadata(args) + rotation = _resolve_rotation(args) try: result = rfapi.create_api_key( @@ -423,6 +545,7 @@ def _create_key(args) -> None: # noqa: ANN001 folder_ids=folder_ids, custom_metadata=None if metadata is _UNSET else metadata, protected=getattr(args, "protected", False), + **rotation, ) except rfapi.RoboflowError as exc: status = getattr(exc, "status_code", None) @@ -442,7 +565,7 @@ def _create_key(args) -> None: # noqa: ANN001 output_api_error( args, exc, - hint="Scopes, folders, and metadata require the Advanced API Keys plan feature.", + hint="Scopes, folders, metadata and the rotation schedule require the Advanced API Keys plan feature.", ) return @@ -478,6 +601,7 @@ def _update_key(args) -> None: # noqa: ANN001 metadata = _resolve_metadata(args) if metadata is not _UNSET: fields["custom_metadata"] = metadata + fields.update(_resolve_rotation(args)) try: result = rfapi.update_api_key(api_key, ws, args.key_id, **fields) @@ -587,3 +711,72 @@ def _revoke_key(args) -> None: # noqa: ANN001 return output(args, result, text=f"Revoked API key '{args.key_id}'.") + + +def _roll_key(args) -> None: # noqa: ANN001 + from roboflow.adapters import rfapi + from roboflow.cli._output import confirm_destructive, output, output_api_error, output_error + + resolved = _resolve_ws_and_key(args) + if not resolved: + return + ws, api_key = resolved + + if not confirm_destructive(args, f"Roll API key '{args.key_id}'? The current key stops working immediately."): + return + + try: + result = rfapi.roll_api_key( + api_key, + ws, + args.key_id, + rotation_due_at=getattr(args, "rotation_due_at", None), + owners=getattr(args, "owner", None) or None, + ) + except rfapi.RoboflowError as exc: + status = getattr(exc, "status_code", None) + if status == 409: + output_error( + args, + str(exc), + hint=( + "The key is protected, is already being rolled, or is disabled while backing a " + "Dedicated Deployment. Unprotect it at app.roboflow.com/settings/api, or retry once " + "the other roll finishes." + ), + exit_code=1, + ) + elif status == 403: + output_api_error( + args, + exc, + hint=( + "Rolling needs the 'api-key:create' and 'api-key:revoke' scopes, and the calling key " + "must hold at least the target key's scopes and folders. --rotation-due-at and --owner " + "require the Advanced API Keys plan feature." + ), + ) + else: + output_api_error( + args, + exc, + not_found_hint=( + f"No active API key with ID '{args.key_id}' in this workspace. " + "Rolled and revoked keys are not listed." + ), + ) + return + + secret = result.get("key", "") + new_key_id = result.get("keyId", "") + name = result.get("name") or args.key_id + + lines = [ + f"Rolled API key '{name}': revoked {args.key_id}, replacement keyId: {new_key_id}", + "", + "WARNING: This is the only time the new secret key will be shown.", + "Save it somewhere secure now and update everything that used the old key.", + "", + f" Key: {secret}", + ] + output(args, result, text="\n".join(lines)) diff --git a/tests/cli/test_api_key.py b/tests/cli/test_api_key.py index fba7a389..d7c100c5 100644 --- a/tests/cli/test_api_key.py +++ b/tests/cli/test_api_key.py @@ -78,6 +78,20 @@ def test_revoke_help(self) -> None: result = runner.invoke(app, ["api-key", "revoke", "--help"]) self.assertEqual(result.exit_code, 0, result.output) + def test_create_and_update_help_list_rotation_flags(self) -> None: + create_help = _strip_ansi(runner.invoke(app, ["api-key", "create", "--help"]).output) + update_help = _strip_ansi(runner.invoke(app, ["api-key", "update", "--help"]).output) + for flag in ("--rotation-due-at", "--owner"): + self.assertIn(flag, create_help) + self.assertIn(flag, update_help) + self.assertIn("--clear-rotation-due-at", update_help) + self.assertIn("--clear-owners", update_help) + + def test_roll_help(self) -> None: + result = runner.invoke(app, ["api-key", "roll", "--help"]) + self.assertEqual(result.exit_code, 0, result.output) + self.assertIn("--rotation-due-at", _strip_ansi(result.output)) + # --------------------------------------------------------------------------- # list @@ -651,6 +665,258 @@ def test_revoke_json(self, _mock_key, _mock_ws, mock_revoke) -> None: self.assertEqual(data["status"], "revoked") +# --------------------------------------------------------------------------- +# rotation schedule on create / update +# --------------------------------------------------------------------------- + + +class TestRotationScheduleFlags(unittest.TestCase): + def _create_args(self, **overrides): # noqa: ANN001, ANN202 + base = dict( + json=False, workspace=None, api_key=None, quiet=False, name="K", scope=None, folder=None, metadata=None + ) + base.update(overrides) + return Namespace(**base) + + def _update_args(self, **overrides): # noqa: ANN001, ANN202 + base = dict( + json=False, workspace=None, api_key=None, quiet=False, key_id="k1", name=None, scope=None, metadata=None + ) + base.update(overrides) + return Namespace(**base) + + @patch("roboflow.adapters.rfapi.create_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_create_forwards_schedule(self, _mock_key, _mock_ws, mock_create) -> None: + mock_create.return_value = {"keyId": "k1", "key": "SECRET"} + from roboflow.cli.handlers.api_key import _create_key + + with patch("builtins.print"): + _create_key(self._create_args(rotation_due_at="2027-10-23", owner=["u1", "u2"])) + self.assertEqual(mock_create.call_args.kwargs["rotation_due_at"], "2027-10-23") + self.assertEqual(mock_create.call_args.kwargs["owners"], ["u1", "u2"]) + + @patch("roboflow.adapters.rfapi.update_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_update_forwards_schedule(self, _mock_key, _mock_ws, mock_update) -> None: + mock_update.return_value = {"apiKey": {"keyId": "k1"}} + from roboflow.cli.handlers.api_key import _update_key + + with patch("builtins.print"): + _update_key(self._update_args(rotation_due_at="2027-10-23", owner=["u1"])) + mock_update.assert_called_once_with("fake-key", "test-ws", "k1", rotation_due_at="2027-10-23", owners=["u1"]) + + @patch("roboflow.adapters.rfapi.update_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_update_clear_flags_send_null_and_empty_list(self, _mock_key, _mock_ws, mock_update) -> None: + mock_update.return_value = {"apiKey": {"keyId": "k1"}} + from roboflow.adapters.rfapi import CLEAR + from roboflow.cli.handlers.api_key import _update_key + + with patch("builtins.print"): + _update_key(self._update_args(clear_rotation_due_at=True, clear_owners=True)) + mock_update.assert_called_once_with("fake-key", "test-ws", "k1", rotation_due_at=CLEAR, owners=[]) + + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_update_set_and_clear_conflicts_exit_1(self, _mock_key, _mock_ws) -> None: + from roboflow.cli.handlers.api_key import _update_key + + for overrides in ( + dict(rotation_due_at="2027-10-23", clear_rotation_due_at=True), + dict(owner=["u1"], clear_owners=True), + ): + with patch("sys.stderr"): + with self.assertRaises(SystemExit) as ctx: + _update_key(self._update_args(**overrides)) + self.assertEqual(ctx.exception.code, 1) + + @patch("roboflow.util.http.requests.post") + def test_create_serializes_schedule(self, mock_post) -> None: + from unittest.mock import MagicMock + + from roboflow.adapters.rfapi import create_api_key + + resp = MagicMock() + resp.ok = True + resp.json.return_value = {"keyId": "k1", "key": "SECRET"} + mock_post.return_value = resp + create_api_key("fake-key", "test-ws", name="K", rotation_due_at="2027-10-23", owners=[]) + body = mock_post.call_args.kwargs["json"] + self.assertEqual(body["rotationDueAt"], "2027-10-23") + self.assertEqual(body["owners"], []) + + @patch("roboflow.util.http.requests.patch") + def test_update_clear_serializes_null_due_date(self, mock_patch) -> None: + from unittest.mock import MagicMock + + from roboflow.adapters.rfapi import CLEAR, update_api_key + + resp = MagicMock() + resp.ok = True + resp.json.return_value = {"apiKey": {"keyId": "k1"}} + mock_patch.return_value = resp + update_api_key("fake-key", "test-ws", "k1", rotation_due_at=CLEAR, owners=[]) + self.assertEqual(mock_patch.call_args.kwargs["json"], {"rotationDueAt": None, "owners": []}) + + +# --------------------------------------------------------------------------- +# roll +# --------------------------------------------------------------------------- + + +def _roll_args(**overrides): # noqa: ANN001, ANN202 + base = dict( + json=False, workspace=None, api_key=None, quiet=False, key_id="k1", rotation_due_at=None, owner=None, yes=True + ) + base.update(overrides) + return Namespace(**base) + + +_ROLLED = { + "keyId": "k2", + "key": "NEWSECRET", + "name": "My Key", + "scopes": None, + "folderIds": [], + "protected": False, + "disabled": False, + "rotationDueAt": None, + "owners": [], + "revokedKeyId": "k1", +} + + +class TestRollApiKey(unittest.TestCase): + @patch("roboflow.adapters.rfapi.roll_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_roll_with_yes_prints_secret_once(self, _mock_key, _mock_ws, mock_roll) -> None: + mock_roll.return_value = dict(_ROLLED) + from roboflow.cli.handlers.api_key import _roll_key + + with patch("builtins.print") as mock_print: + _roll_key(_roll_args()) + mock_roll.assert_called_once_with("fake-key", "test-ws", "k1", rotation_due_at=None, owners=None) + printed = mock_print.call_args[0][0] + self.assertIn("NEWSECRET", printed) + self.assertIn("k2", printed) + self.assertIn("revoked k1", printed) + + @patch("roboflow.adapters.rfapi.roll_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_roll_forwards_schedule(self, _mock_key, _mock_ws, mock_roll) -> None: + mock_roll.return_value = dict(_ROLLED) + from roboflow.cli.handlers.api_key import _roll_key + + with patch("builtins.print"): + _roll_key(_roll_args(rotation_due_at="2027-10-22", owner=["u1", "u2"])) + mock_roll.assert_called_once_with( + "fake-key", "test-ws", "k1", rotation_due_at="2027-10-22", owners=["u1", "u2"] + ) + + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_roll_without_yes_no_tty_exits(self, _mock_key, _mock_ws) -> None: + from roboflow.cli.handlers.api_key import _roll_key + + with patch("sys.stdin.isatty", return_value=False): + with self.assertRaises(SystemExit) as ctx: + _roll_key(_roll_args(yes=False)) + self.assertEqual(ctx.exception.code, 1) + + @patch("roboflow.adapters.rfapi.roll_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_roll_json(self, _mock_key, _mock_ws, mock_roll) -> None: + mock_roll.return_value = dict(_ROLLED) + from roboflow.cli.handlers.api_key import _roll_key + + with patch("builtins.print") as mock_print: + _roll_key(_roll_args(json=True)) + data = json.loads(mock_print.call_args[0][0]) + self.assertEqual(data["key"], "NEWSECRET") + self.assertEqual(data["revokedKeyId"], "k1") + + @patch("roboflow.adapters.rfapi.roll_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_roll_protected_409_exits_1(self, _mock_key, _mock_ws, mock_roll) -> None: + from roboflow.adapters.rfapi import RoboflowError + from roboflow.cli.handlers.api_key import _roll_key + + mock_roll.side_effect = RoboflowError("Cannot roll a protected API key", status_code=409) + with patch("sys.stderr"): + with self.assertRaises(SystemExit) as ctx: + _roll_key(_roll_args()) + self.assertEqual(ctx.exception.code, 1) + + @patch("roboflow.adapters.rfapi.roll_api_key") + @patch("roboflow.cli._resolver.resolve_default_workspace", return_value="test-ws") + @patch("roboflow.config.load_roboflow_api_key", return_value="fake-key") + def test_roll_missing_exits_3(self, _mock_key, _mock_ws, mock_roll) -> None: + from roboflow.adapters.rfapi import RoboflowError + from roboflow.cli.handlers.api_key import _roll_key + + mock_roll.side_effect = RoboflowError("not found", status_code=404) + with patch("sys.stderr") as mock_stderr: + with self.assertRaises(SystemExit) as ctx: + _roll_key(_roll_args(key_id="missing")) + self.assertEqual(ctx.exception.code, 3) + written = "".join(str(call.args[0]) for call in mock_stderr.write.call_args_list) + self.assertIn("Rolled and revoked keys are not listed", written) + + +class TestRollApiKeyAdapter(unittest.TestCase): + def _ok_response(self, payload=None): # noqa: ANN001, ANN202 + from unittest.mock import MagicMock + + resp = MagicMock() + resp.ok = True + resp.status_code = 200 + resp.json.return_value = payload or dict(_ROLLED) + return resp + + @patch("roboflow.util.http.requests.post") + def test_roll_posts_to_roll_path_without_body(self, mock_post) -> None: + from roboflow.adapters.rfapi import roll_api_key + + mock_post.return_value = self._ok_response() + result = roll_api_key("fake-key", "test-ws", "k 1") + url = mock_post.call_args.args[0] + self.assertTrue(url.endswith("/test-ws/api-keys/k%201/roll"), url) + self.assertEqual(mock_post.call_args.kwargs["params"], {"api_key": "fake-key"}) + self.assertIsNone(mock_post.call_args.kwargs["json"]) + self.assertEqual(result["key"], "NEWSECRET") + + @patch("roboflow.util.http.requests.post") + def test_roll_forwards_replacement_schedule(self, mock_post) -> None: + from roboflow.adapters.rfapi import roll_api_key + + mock_post.return_value = self._ok_response() + roll_api_key("fake-key", "test-ws", "k1", rotation_due_at="2027-10-22", owners=["u1"]) + self.assertEqual(mock_post.call_args.kwargs["json"], {"rotationDueAt": "2027-10-22", "owners": ["u1"]}) + + @patch("roboflow.util.http.requests.post") + def test_roll_attaches_status_code(self, mock_post) -> None: + from unittest.mock import MagicMock + + from roboflow.adapters.rfapi import RoboflowError, roll_api_key + + resp = MagicMock() + resp.ok = False + resp.status_code = 409 + resp.text = "already rolled" + mock_post.return_value = resp + with self.assertRaises(RoboflowError) as ctx: + roll_api_key("fake-key", "test-ws", "k1") + self.assertEqual(ctx.exception.status_code, 409) + + # --------------------------------------------------------------------------- # status-code propagation / exit-code consistency # ---------------------------------------------------------------------------