From 9deec72245ef5cbe04988e33aa9e7a5939ae07ed Mon Sep 17 00:00:00 2001 From: George-iam Date: Mon, 2 Mar 2026 14:46:06 +0000 Subject: [PATCH] feat: add enterprise service-account lifecycle helpers Expose service-account create/list/get and key create/revoke methods in the Python SDK so enterprise admin automation can manage scoped machine credentials through stable client APIs. Add focused client tests for endpoint wiring and response handling. Made-with: Cursor --- axme_sdk/client.py | 80 +++++++++++++++++++++++++++++++++++++++++ tests/test_client.py | 86 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 166 insertions(+) diff --git a/axme_sdk/client.py b/axme_sdk/client.py index bfa90af..bd1d2f5 100644 --- a/axme_sdk/client.py +++ b/axme_sdk/client.py @@ -557,6 +557,86 @@ def update_user_profile( retryable=idempotency_key is not None, ) + def create_service_account( + self, + payload: dict[str, Any], + *, + idempotency_key: str | None = None, + trace_id: str | None = None, + ) -> dict[str, Any]: + return self._request_json( + "POST", + "/v1/service-accounts", + json_body=payload, + idempotency_key=idempotency_key, + trace_id=trace_id, + retryable=idempotency_key is not None, + ) + + def list_service_accounts( + self, + *, + org_id: str, + workspace_id: str | None = None, + trace_id: str | None = None, + ) -> dict[str, Any]: + params: dict[str, str] = {"org_id": org_id} + if workspace_id is not None: + params["workspace_id"] = workspace_id + return self._request_json( + "GET", + "/v1/service-accounts", + params=params, + trace_id=trace_id, + retryable=True, + ) + + def get_service_account( + self, + service_account_id: str, + *, + trace_id: str | None = None, + ) -> dict[str, Any]: + return self._request_json( + "GET", + f"/v1/service-accounts/{service_account_id}", + trace_id=trace_id, + retryable=True, + ) + + def create_service_account_key( + self, + service_account_id: str, + payload: dict[str, Any], + *, + idempotency_key: str | None = None, + trace_id: str | None = None, + ) -> dict[str, Any]: + return self._request_json( + "POST", + f"/v1/service-accounts/{service_account_id}/keys", + json_body=payload, + idempotency_key=idempotency_key, + trace_id=trace_id, + retryable=idempotency_key is not None, + ) + + def revoke_service_account_key( + self, + service_account_id: str, + key_id: str, + *, + idempotency_key: str | None = None, + trace_id: str | None = None, + ) -> dict[str, Any]: + return self._request_json( + "POST", + f"/v1/service-accounts/{service_account_id}/keys/{key_id}/revoke", + idempotency_key=idempotency_key, + trace_id=trace_id, + retryable=idempotency_key is not None, + ) + def upsert_webhook_subscription( self, payload: dict[str, Any], diff --git a/tests/test_client.py b/tests/test_client.py index 3f59aad..fd01247 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -984,6 +984,92 @@ def handler(request: httpx.Request) -> httpx.Response: assert exc_info.value.body["message"] == "too many" +def test_create_and_list_service_accounts_success() -> None: + org_id = "aaaaaaaa-aaaa-4aaa-8aaa-aaaaaaaaaaaa" + workspace_id = "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" + service_account_id = "sa_123" + + def handler(request: httpx.Request) -> httpx.Response: + if request.method == "POST": + assert request.url.path == "/v1/service-accounts" + body = json.loads(request.read().decode("utf-8")) + assert body["org_id"] == org_id + assert body["workspace_id"] == workspace_id + return httpx.Response( + 200, + json={ + "ok": True, + "service_account": { + "service_account_id": service_account_id, + "org_id": org_id, + "workspace_id": workspace_id, + }, + }, + ) + assert request.method == "GET" + assert request.url.path == "/v1/service-accounts" + assert request.url.params.get("org_id") == org_id + assert request.url.params.get("workspace_id") == workspace_id + return httpx.Response(200, json={"ok": True, "service_accounts": [{"service_account_id": service_account_id}]}) + + client = _client(handler) + created = client.create_service_account( + { + "org_id": org_id, + "workspace_id": workspace_id, + "name": "sdk-runner", + "created_by_actor_id": "actor_sdk", + } + ) + assert created["service_account"]["service_account_id"] == service_account_id + listed = client.list_service_accounts(org_id=org_id, workspace_id=workspace_id) + assert listed["service_accounts"][0]["service_account_id"] == service_account_id + + +def test_get_service_account_success() -> None: + service_account_id = "sa_abc" + + def handler(request: httpx.Request) -> httpx.Response: + assert request.method == "GET" + assert request.url.path == f"/v1/service-accounts/{service_account_id}" + return httpx.Response(200, json={"ok": True, "service_account": {"service_account_id": service_account_id}}) + + client = _client(handler) + fetched = client.get_service_account(service_account_id) + assert fetched["service_account"]["service_account_id"] == service_account_id + + +def test_create_and_revoke_service_account_key_success() -> None: + service_account_id = "sa_abc" + key_id = "sak_abc" + + def handler(request: httpx.Request) -> httpx.Response: + if request.url.path.endswith("/keys"): + assert request.method == "POST" + assert request.url.path == f"/v1/service-accounts/{service_account_id}/keys" + return httpx.Response( + 200, + json={ + "ok": True, + "key": { + "key_id": key_id, + "service_account_id": service_account_id, + "status": "active", + "token": "axme_sa_token", + }, + }, + ) + assert request.method == "POST" + assert request.url.path == f"/v1/service-accounts/{service_account_id}/keys/{key_id}/revoke" + return httpx.Response(200, json={"ok": True, "key": {"key_id": key_id, "status": "revoked"}}) + + client = _client(handler) + created = client.create_service_account_key(service_account_id, {"created_by_actor_id": "actor_sdk"}) + assert created["key"]["key_id"] == key_id + revoked = client.revoke_service_account_key(service_account_id, key_id) + assert revoked["key"]["status"] == "revoked" + + def test_upsert_webhook_subscription_success() -> None: subscription = _webhook_subscription_payload() request_payload = {