From 2c701571f8ebf0e6ddbd43a3089934c7bc75e56a Mon Sep 17 00:00:00 2001 From: dorsha Date: Fri, 10 Jul 2026 18:42:16 +0300 Subject: [PATCH 1/3] feat(mgmt): add outbound SCIM configuration management wrapper Adds descope.management.OutboundSCIM (sync + async) with create_configuration, update_configuration, delete_configuration, load_configuration, load_all_configurations, and set_enabled. Related: descope/etc#15987 Co-Authored-By: Claude Opus 4.7 --- descope/management/_outbound_scim_base.py | 36 +++ descope/management/common.py | 8 + descope/management/outbound_scim.py | 148 ++++++++++++ descope/management/outbound_scim_async.py | 150 ++++++++++++ descope/mgmt.py | 7 + descope/mgmt_async.py | 7 + tests/management/test_outbound_scim.py | 276 ++++++++++++++++++++++ 7 files changed, 632 insertions(+) create mode 100644 descope/management/_outbound_scim_base.py create mode 100644 descope/management/outbound_scim.py create mode 100644 descope/management/outbound_scim_async.py create mode 100644 tests/management/test_outbound_scim.py diff --git a/descope/management/_outbound_scim_base.py b/descope/management/_outbound_scim_base.py new file mode 100644 index 000000000..2e4513c84 --- /dev/null +++ b/descope/management/_outbound_scim_base.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from typing import Any, Optional + + +class OutboundSCIMBase: + @staticmethod + def _compose_create_body( + name: str, + app_id: str, + configuration: Optional[dict] = None, + ) -> dict: + body: dict[str, Any] = { + "name": name, + "appId": app_id, + } + if configuration is not None: + body["configuration"] = configuration + return body + + @staticmethod + def _compose_update_body( + id: str, + version: int, + configuration: Optional[dict] = None, + name: Optional[str] = None, + ) -> dict: + body: dict[str, Any] = { + "id": id, + "version": version, + } + if name is not None: + body["name"] = name + if configuration is not None: + body["configuration"] = configuration + return body diff --git a/descope/management/common.py b/descope/management/common.py index 9c2df5ea5..140cf9409 100644 --- a/descope/management/common.py +++ b/descope/management/common.py @@ -128,6 +128,14 @@ class MgmtV1: outbound_application_batch_upload_user_tokens_path = "/v1/mgmt/outbound/app/user/oauthtoken/batch/upload" outbound_application_batch_upload_tenant_tokens_path = "/v1/mgmt/outbound/app/tenant/oauthtoken/batch/upload" + # outbound scim + outbound_scim_create_path = "/v1/mgmt/outbound/scim/create" + outbound_scim_update_path = "/v1/mgmt/outbound/scim/update" + outbound_scim_delete_path = "/v1/mgmt/outbound/scim/delete" + outbound_scim_load_path = "/v1/mgmt/outbound/scim" + outbound_scim_load_all_path = "/v1/mgmt/outbound/scim" + outbound_scim_set_enabled_path = "/v1/mgmt/outbound/scim/enabled/set" + # engine engine_create_path = "/v1/mgmt/engine/create" engine_update_path = "/v1/mgmt/engine/update" diff --git a/descope/management/outbound_scim.py b/descope/management/outbound_scim.py new file mode 100644 index 000000000..9bf88234e --- /dev/null +++ b/descope/management/outbound_scim.py @@ -0,0 +1,148 @@ +from __future__ import annotations + +from typing import Optional + +from descope._http_base import HTTPBase +from descope.management._outbound_scim_base import OutboundSCIMBase +from descope.management.common import MgmtV1 + + +class OutboundSCIM(OutboundSCIMBase, HTTPBase): + def create_configuration( + self, + name: str, + app_id: str, + configuration: Optional[dict] = None, + ) -> dict: + """ + Create a new outbound SCIM configuration. The configuration ID is provisioned + automatically by Descope and returned in the response. + + Args: + name (str): The outbound SCIM configuration's name. + app_id (str): The ID of the outbound application this SCIM configuration binds to. + configuration (dict): Optional provider-specific SCIM configuration dictionary. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , "appId": , + "configuration": {...}, "enabled": , + "lastExportTime": , "lastProcessingTime": , + "failures": , "version": }} + + Raise: + AuthException: raised if create operation fails + """ + response = self._http.post( + MgmtV1.outbound_scim_create_path, + body=OutboundSCIMBase._compose_create_body(name, app_id, configuration), + ) + return response.json() + + def update_configuration( + self, + id: str, + version: int, + configuration: Optional[dict] = None, + name: Optional[str] = None, + ) -> dict: + """ + Update an existing outbound SCIM configuration. The version must match the + currently stored version — otherwise the update is rejected as a conflict. + + Args: + id (str): The ID of the outbound SCIM configuration to update. + version (int): The currently stored version, used for optimistic concurrency. + configuration (dict): Optional updated provider-specific SCIM configuration. + name (str): Optional updated configuration name. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , ...}} + + Raise: + AuthException: raised if update operation fails + """ + response = self._http.post( + MgmtV1.outbound_scim_update_path, + body=OutboundSCIMBase._compose_update_body(id, version, configuration, name), + ) + return response.json() + + def delete_configuration( + self, + id: str, + ): + """ + Delete an existing outbound SCIM configuration. IMPORTANT: This action is + irreversible. Use carefully. + + Args: + id (str): The ID of the outbound SCIM configuration to delete. + + Raise: + AuthException: raised if deletion operation fails + """ + self._http.post(MgmtV1.outbound_scim_delete_path, body={"id": id}) + + def load_configuration( + self, + id: str, + ) -> dict: + """ + Load an outbound SCIM configuration by ID. + + Args: + id (str): The ID of the outbound SCIM configuration to load. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , "appId": , + "configuration": {...}, "enabled": , + "lastExportTime": , "lastProcessingTime": , + "failures": , "version": }} + + Raise: + AuthException: raised if load operation fails + """ + response = self._http.get(f"{MgmtV1.outbound_scim_load_path}/{id}") + return response.json() + + def load_all_configurations(self) -> dict: + """ + Load all outbound SCIM configurations for the project. + + Return value (dict): + Return dict in the format + {"configurations": [{"id": , "name": , ...}, ...]} + + Raise: + AuthException: raised if load operation fails + """ + response = self._http.get(MgmtV1.outbound_scim_load_all_path) + return response.json() + + def set_enabled( + self, + id: str, + enabled: bool, + ) -> dict: + """ + Enable or disable an outbound SCIM configuration. + + Args: + id (str): The ID of the outbound SCIM configuration to update. + enabled (bool): Whether the SCIM configuration should be enabled. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , "enabled": , ...}} + + Raise: + AuthException: raised if the operation fails + """ + response = self._http.post( + MgmtV1.outbound_scim_set_enabled_path, + body={"id": id, "enabled": enabled}, + ) + return response.json() diff --git a/descope/management/outbound_scim_async.py b/descope/management/outbound_scim_async.py new file mode 100644 index 000000000..0fc7b29b3 --- /dev/null +++ b/descope/management/outbound_scim_async.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +from typing import Optional + +from descope._http_base import AsyncHTTPBase +from descope.management._outbound_scim_base import OutboundSCIMBase +from descope.management.common import MgmtV1 + + +class OutboundSCIMAsync(OutboundSCIMBase, AsyncHTTPBase): + """Async counterpart of OutboundSCIM — all HTTP calls are coroutines.""" + + async def create_configuration( + self, + name: str, + app_id: str, + configuration: Optional[dict] = None, + ) -> dict: + """ + Create a new outbound SCIM configuration. The configuration ID is provisioned + automatically by Descope and returned in the response. + + Args: + name (str): The outbound SCIM configuration's name. + app_id (str): The ID of the outbound application this SCIM configuration binds to. + configuration (dict): Optional provider-specific SCIM configuration dictionary. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , "appId": , + "configuration": {...}, "enabled": , + "lastExportTime": , "lastProcessingTime": , + "failures": , "version": }} + + Raise: + AuthException: raised if create operation fails + """ + response = await self._http.post( + MgmtV1.outbound_scim_create_path, + body=OutboundSCIMBase._compose_create_body(name, app_id, configuration), + ) + return response.json() + + async def update_configuration( + self, + id: str, + version: int, + configuration: Optional[dict] = None, + name: Optional[str] = None, + ) -> dict: + """ + Update an existing outbound SCIM configuration. The version must match the + currently stored version — otherwise the update is rejected as a conflict. + + Args: + id (str): The ID of the outbound SCIM configuration to update. + version (int): The currently stored version, used for optimistic concurrency. + configuration (dict): Optional updated provider-specific SCIM configuration. + name (str): Optional updated configuration name. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , ...}} + + Raise: + AuthException: raised if update operation fails + """ + response = await self._http.post( + MgmtV1.outbound_scim_update_path, + body=OutboundSCIMBase._compose_update_body(id, version, configuration, name), + ) + return response.json() + + async def delete_configuration( + self, + id: str, + ): + """ + Delete an existing outbound SCIM configuration. IMPORTANT: This action is + irreversible. Use carefully. + + Args: + id (str): The ID of the outbound SCIM configuration to delete. + + Raise: + AuthException: raised if deletion operation fails + """ + await self._http.post(MgmtV1.outbound_scim_delete_path, body={"id": id}) + + async def load_configuration( + self, + id: str, + ) -> dict: + """ + Load an outbound SCIM configuration by ID. + + Args: + id (str): The ID of the outbound SCIM configuration to load. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , "appId": , + "configuration": {...}, "enabled": , + "lastExportTime": , "lastProcessingTime": , + "failures": , "version": }} + + Raise: + AuthException: raised if load operation fails + """ + response = await self._http.get(f"{MgmtV1.outbound_scim_load_path}/{id}") + return response.json() + + async def load_all_configurations(self) -> dict: + """ + Load all outbound SCIM configurations for the project. + + Return value (dict): + Return dict in the format + {"configurations": [{"id": , "name": , ...}, ...]} + + Raise: + AuthException: raised if load operation fails + """ + response = await self._http.get(MgmtV1.outbound_scim_load_all_path) + return response.json() + + async def set_enabled( + self, + id: str, + enabled: bool, + ) -> dict: + """ + Enable or disable an outbound SCIM configuration. + + Args: + id (str): The ID of the outbound SCIM configuration to update. + enabled (bool): Whether the SCIM configuration should be enabled. + + Return value (dict): + Return dict in the format + {"configuration": {"id": , "name": , "enabled": , ...}} + + Raise: + AuthException: raised if the operation fails + """ + response = await self._http.post( + MgmtV1.outbound_scim_set_enabled_path, + body={"id": id, "enabled": enabled}, + ) + return response.json() diff --git a/descope/mgmt.py b/descope/mgmt.py index cab52305f..97ed4375b 100644 --- a/descope/mgmt.py +++ b/descope/mgmt.py @@ -20,6 +20,7 @@ OutboundApplication, OutboundApplicationByToken, ) +from descope.management.outbound_scim import OutboundSCIM from descope.management.password import Password from descope.management.permission import Permission from descope.management.project import Project @@ -59,6 +60,7 @@ def __init__(self, http_client: HTTPClient, auth: Auth, fga_cache_url: Optional[ self._management_key = ManagementKey(http_client) self._outbound_application = OutboundApplication(http_client) self._outbound_application_by_token = OutboundApplicationByToken(http_client) + self._outbound_scim = OutboundSCIM(http_client) self._password = Password(http_client) self._permission = Permission(http_client) self._project = Project(http_client) @@ -163,6 +165,11 @@ def outbound_application_by_token(self): # No management key check for outbound_app_token (as authentication for those methods is done by inbound app token) return self._outbound_application_by_token + @property + def outbound_scim(self): + self._ensure_management_key("outbound_scim") + return self._outbound_scim + @property def descoper(self): self._ensure_management_key("descoper") diff --git a/descope/mgmt_async.py b/descope/mgmt_async.py index 790bb7d3d..78a896f4f 100644 --- a/descope/mgmt_async.py +++ b/descope/mgmt_async.py @@ -20,6 +20,7 @@ OutboundApplicationAsync, OutboundApplicationByTokenAsync, ) +from descope.management.outbound_scim_async import OutboundSCIMAsync from descope.management.password_async import PasswordAsync from descope.management.permission_async import PermissionAsync from descope.management.project_async import ProjectAsync @@ -61,6 +62,7 @@ def __init__(self, http_client: HTTPClientAsync, auth: AuthAsync, fga_cache_url: self._management_key = ManagementKeyAsync(http_client) self._outbound_application = OutboundApplicationAsync(http_client) self._outbound_application_by_token = OutboundApplicationByTokenAsync(http_client) + self._outbound_scim = OutboundSCIMAsync(http_client) self._password = PasswordAsync(http_client) self._permission = PermissionAsync(http_client) self._project = ProjectAsync(http_client) @@ -165,6 +167,11 @@ def outbound_application_by_token(self) -> OutboundApplicationByTokenAsync: # No management key check for outbound_app_token (as authentication for those methods is done by inbound app token) return self._outbound_application_by_token + @property + def outbound_scim(self) -> OutboundSCIMAsync: + self._ensure_management_key("outbound_scim") + return self._outbound_scim + @property def descoper(self) -> DescoperAsync: self._ensure_management_key("descoper") diff --git a/tests/management/test_outbound_scim.py b/tests/management/test_outbound_scim.py new file mode 100644 index 000000000..e9854cf3d --- /dev/null +++ b/tests/management/test_outbound_scim.py @@ -0,0 +1,276 @@ +import pytest + +from descope import AuthException +from descope.management.common import MgmtV1 +from descope.management.outbound_scim import OutboundSCIM +from tests.common import DEFAULT_BASE_URL, default_headers +from tests.conftest import PROJECT_ID, assert_http_called, make_response +from tests.testutils import PUBLIC_KEY_DICT + +CONFIG_RESPONSE = { + "configuration": { + "id": "scim1", + "name": "Test SCIM", + "appId": "app1", + "configuration": {"baseUrl": "https://scim.example.com"}, + "enabled": True, + "lastExportTime": 1719571200, + "lastProcessingTime": 1719571300, + "failures": 0, + "version": 3, + } +} + +CONFIGS_RESPONSE = { + "configurations": [ + {"id": "scim1", "name": "Test SCIM 1", "appId": "app1", "version": 1}, + {"id": "scim2", "name": "Test SCIM 2", "appId": "app2", "version": 2}, + ] +} + +MGMT_HEADERS = { + **default_headers, + "Authorization": f"Bearer {PROJECT_ID}:key", + "x-descope-project-id": PROJECT_ID, +} + + +class TestOutboundSCIM: + async def test_create_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke( + client.mgmt.outbound_scim.create_configuration( + "Test SCIM", + "app1", + {"baseUrl": "https://scim.example.com"}, + ) + ) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_create_path}", + headers=MGMT_HEADERS, + params=None, + json={ + "name": "Test SCIM", + "appId": "app1", + "configuration": {"baseUrl": "https://scim.example.com"}, + }, + follow_redirects=False, + ) + + async def test_create_configuration_minimal_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke(client.mgmt.outbound_scim.create_configuration("Test SCIM", "app1")) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_create_path}", + headers=MGMT_HEADERS, + params=None, + json={"name": "Test SCIM", "appId": "app1"}, + follow_redirects=False, + ) + + async def test_create_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.create_configuration("Test SCIM", "app1")) + + async def test_update_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke( + client.mgmt.outbound_scim.update_configuration( + "scim1", + 3, + {"baseUrl": "https://scim.example.com"}, + "Updated Name", + ) + ) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_update_path}", + headers=MGMT_HEADERS, + params=None, + json={ + "id": "scim1", + "version": 3, + "name": "Updated Name", + "configuration": {"baseUrl": "https://scim.example.com"}, + }, + follow_redirects=False, + ) + + async def test_update_configuration_without_name_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + await client.invoke( + client.mgmt.outbound_scim.update_configuration("scim1", 3, {"baseUrl": "https://scim.example.com"}) + ) + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_update_path}", + headers=MGMT_HEADERS, + params=None, + json={ + "id": "scim1", + "version": 3, + "configuration": {"baseUrl": "https://scim.example.com"}, + }, + follow_redirects=False, + ) + + async def test_update_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.update_configuration("scim1", 3, {})) + + async def test_delete_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response({})) as mock_post: + await client.invoke(client.mgmt.outbound_scim.delete_configuration("scim1")) + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_delete_path}", + headers=MGMT_HEADERS, + params=None, + json={"id": "scim1"}, + follow_redirects=False, + ) + + async def test_delete_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.delete_configuration("scim1")) + + async def test_load_configuration_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_get(make_response(CONFIG_RESPONSE)) as mock_get: + response = await client.invoke(client.mgmt.outbound_scim.load_configuration("scim1")) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_get, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_load_path}/scim1", + headers=MGMT_HEADERS, + params=None, + follow_redirects=True, + ) + + async def test_load_configuration_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_get(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.load_configuration("scim1")) + + async def test_load_all_configurations_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_get(make_response(CONFIGS_RESPONSE)) as mock_get: + response = await client.invoke(client.mgmt.outbound_scim.load_all_configurations()) + assert response == CONFIGS_RESPONSE + assert_http_called( + mock_get, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_load_all_path}", + headers=MGMT_HEADERS, + params=None, + follow_redirects=True, + ) + + async def test_load_all_configurations_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_get(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.load_all_configurations()) + + async def test_set_enabled_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + response = await client.invoke(client.mgmt.outbound_scim.set_enabled("scim1", True)) + assert response == CONFIG_RESPONSE + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_set_enabled_path}", + headers=MGMT_HEADERS, + params=None, + json={"id": "scim1", "enabled": True}, + follow_redirects=False, + ) + + async def test_set_enabled_false_success(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: + await client.invoke(client.mgmt.outbound_scim.set_enabled("scim1", False)) + assert_http_called( + mock_post, + client.mode, + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_set_enabled_path}", + headers=MGMT_HEADERS, + params=None, + json={"id": "scim1", "enabled": False}, + follow_redirects=False, + ) + + async def test_set_enabled_failure(self, client_factory): + client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") + + with client.mock_mgmt_post(make_response(status=500)): + with pytest.raises(AuthException): + await client.invoke(client.mgmt.outbound_scim.set_enabled("scim1", True)) + + def test_compose_create_body(self): + body = OutboundSCIM._compose_create_body("Test SCIM", "app1", {"baseUrl": "https://scim.example.com"}) + assert body == { + "name": "Test SCIM", + "appId": "app1", + "configuration": {"baseUrl": "https://scim.example.com"}, + } + + def test_compose_create_body_without_configuration(self): + body = OutboundSCIM._compose_create_body("Test SCIM", "app1") + assert body == {"name": "Test SCIM", "appId": "app1"} + + def test_compose_update_body(self): + body = OutboundSCIM._compose_update_body( + "scim1", + 5, + {"baseUrl": "https://scim.example.com"}, + "New Name", + ) + assert body == { + "id": "scim1", + "version": 5, + "name": "New Name", + "configuration": {"baseUrl": "https://scim.example.com"}, + } + + def test_compose_update_body_without_optionals(self): + body = OutboundSCIM._compose_update_body("scim1", 5) + assert body == {"id": "scim1", "version": 5} From d07ac7233614232beca4f2ed1371ed8223861d29 Mon Sep 17 00:00:00 2001 From: dorsha Date: Wed, 15 Jul 2026 16:20:16 +0300 Subject: [PATCH 2/3] refactor(mgmt): drop load_all_configurations from OutboundSCIM Co-Authored-By: Claude Opus 4.7 --- descope/management/common.py | 1 - descope/management/outbound_scim.py | 14 ----------- descope/management/outbound_scim_async.py | 14 ----------- tests/management/test_outbound_scim.py | 29 ----------------------- 4 files changed, 58 deletions(-) diff --git a/descope/management/common.py b/descope/management/common.py index 140cf9409..0a6466dcf 100644 --- a/descope/management/common.py +++ b/descope/management/common.py @@ -133,7 +133,6 @@ class MgmtV1: outbound_scim_update_path = "/v1/mgmt/outbound/scim/update" outbound_scim_delete_path = "/v1/mgmt/outbound/scim/delete" outbound_scim_load_path = "/v1/mgmt/outbound/scim" - outbound_scim_load_all_path = "/v1/mgmt/outbound/scim" outbound_scim_set_enabled_path = "/v1/mgmt/outbound/scim/enabled/set" # engine diff --git a/descope/management/outbound_scim.py b/descope/management/outbound_scim.py index 9bf88234e..2c74ece0c 100644 --- a/descope/management/outbound_scim.py +++ b/descope/management/outbound_scim.py @@ -108,20 +108,6 @@ def load_configuration( response = self._http.get(f"{MgmtV1.outbound_scim_load_path}/{id}") return response.json() - def load_all_configurations(self) -> dict: - """ - Load all outbound SCIM configurations for the project. - - Return value (dict): - Return dict in the format - {"configurations": [{"id": , "name": , ...}, ...]} - - Raise: - AuthException: raised if load operation fails - """ - response = self._http.get(MgmtV1.outbound_scim_load_all_path) - return response.json() - def set_enabled( self, id: str, diff --git a/descope/management/outbound_scim_async.py b/descope/management/outbound_scim_async.py index 0fc7b29b3..e08b82496 100644 --- a/descope/management/outbound_scim_async.py +++ b/descope/management/outbound_scim_async.py @@ -110,20 +110,6 @@ async def load_configuration( response = await self._http.get(f"{MgmtV1.outbound_scim_load_path}/{id}") return response.json() - async def load_all_configurations(self) -> dict: - """ - Load all outbound SCIM configurations for the project. - - Return value (dict): - Return dict in the format - {"configurations": [{"id": , "name": , ...}, ...]} - - Raise: - AuthException: raised if load operation fails - """ - response = await self._http.get(MgmtV1.outbound_scim_load_all_path) - return response.json() - async def set_enabled( self, id: str, diff --git a/tests/management/test_outbound_scim.py b/tests/management/test_outbound_scim.py index e9854cf3d..8eea3c563 100644 --- a/tests/management/test_outbound_scim.py +++ b/tests/management/test_outbound_scim.py @@ -21,13 +21,6 @@ } } -CONFIGS_RESPONSE = { - "configurations": [ - {"id": "scim1", "name": "Test SCIM 1", "appId": "app1", "version": 1}, - {"id": "scim2", "name": "Test SCIM 2", "appId": "app2", "version": 2}, - ] -} - MGMT_HEADERS = { **default_headers, "Authorization": f"Bearer {PROJECT_ID}:key", @@ -185,28 +178,6 @@ async def test_load_configuration_failure(self, client_factory): with pytest.raises(AuthException): await client.invoke(client.mgmt.outbound_scim.load_configuration("scim1")) - async def test_load_all_configurations_success(self, client_factory): - client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") - - with client.mock_mgmt_get(make_response(CONFIGS_RESPONSE)) as mock_get: - response = await client.invoke(client.mgmt.outbound_scim.load_all_configurations()) - assert response == CONFIGS_RESPONSE - assert_http_called( - mock_get, - client.mode, - f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_load_all_path}", - headers=MGMT_HEADERS, - params=None, - follow_redirects=True, - ) - - async def test_load_all_configurations_failure(self, client_factory): - client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") - - with client.mock_mgmt_get(make_response(status=500)): - with pytest.raises(AuthException): - await client.invoke(client.mgmt.outbound_scim.load_all_configurations()) - async def test_set_enabled_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") From 86a4c276ac15e58c0b1b606a644facd49762cdc2 Mon Sep 17 00:00:00 2001 From: dorsha Date: Wed, 15 Jul 2026 20:54:22 +0300 Subject: [PATCH 3/3] refactor(mgmt-scim): appId-centric OutboundSCIM surface Backend PR #1747 review round 3: drop id + name from OutboundSCIMConfiguration and identify the SCIM configuration by the federated SSO app id end-to-end. Create/Update/Delete/Load/SetEnabled all take app_id now; the connector name is derived server-side from the app. Co-Authored-By: Claude Opus 4.7 --- descope/management/_outbound_scim_base.py | 9 +-- descope/management/outbound_scim.py | 70 +++++++++++----------- descope/management/outbound_scim_async.py | 70 +++++++++++----------- tests/management/test_outbound_scim.py | 71 +++++++++-------------- 4 files changed, 98 insertions(+), 122 deletions(-) diff --git a/descope/management/_outbound_scim_base.py b/descope/management/_outbound_scim_base.py index 2e4513c84..4a36d692e 100644 --- a/descope/management/_outbound_scim_base.py +++ b/descope/management/_outbound_scim_base.py @@ -6,12 +6,10 @@ class OutboundSCIMBase: @staticmethod def _compose_create_body( - name: str, app_id: str, configuration: Optional[dict] = None, ) -> dict: body: dict[str, Any] = { - "name": name, "appId": app_id, } if configuration is not None: @@ -20,17 +18,14 @@ def _compose_create_body( @staticmethod def _compose_update_body( - id: str, + app_id: str, version: int, configuration: Optional[dict] = None, - name: Optional[str] = None, ) -> dict: body: dict[str, Any] = { - "id": id, + "appId": app_id, "version": version, } - if name is not None: - body["name"] = name if configuration is not None: body["configuration"] = configuration return body diff --git a/descope/management/outbound_scim.py b/descope/management/outbound_scim.py index 2c74ece0c..62841c4ba 100644 --- a/descope/management/outbound_scim.py +++ b/descope/management/outbound_scim.py @@ -10,125 +10,123 @@ class OutboundSCIM(OutboundSCIMBase, HTTPBase): def create_configuration( self, - name: str, app_id: str, configuration: Optional[dict] = None, ) -> dict: """ - Create a new outbound SCIM configuration. The configuration ID is provisioned - automatically by Descope and returned in the response. + Create a new outbound SCIM configuration on the federated SSO application + identified by app_id. The connector name is derived server-side from the app. Args: - name (str): The outbound SCIM configuration's name. - app_id (str): The ID of the outbound application this SCIM configuration binds to. + app_id (str): The federated SSO application id this SCIM configuration binds to. configuration (dict): Optional provider-specific SCIM configuration dictionary. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , "appId": , - "configuration": {...}, "enabled": , - "lastExportTime": , "lastProcessingTime": , - "failures": , "version": }} + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} Raise: AuthException: raised if create operation fails """ response = self._http.post( MgmtV1.outbound_scim_create_path, - body=OutboundSCIMBase._compose_create_body(name, app_id, configuration), + body=OutboundSCIMBase._compose_create_body(app_id, configuration), ) return response.json() def update_configuration( self, - id: str, + app_id: str, version: int, configuration: Optional[dict] = None, - name: Optional[str] = None, ) -> dict: """ - Update an existing outbound SCIM configuration. The version must match the - currently stored version — otherwise the update is rejected as a conflict. + Update the outbound SCIM configuration attached to the given federated SSO app. + The version must match the currently stored version — otherwise the update is + rejected as a conflict. Args: - id (str): The ID of the outbound SCIM configuration to update. + app_id (str): The federated SSO application id. version (int): The currently stored version, used for optimistic concurrency. configuration (dict): Optional updated provider-specific SCIM configuration. - name (str): Optional updated configuration name. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , ...}} + {"configuration": {"appId": , ...}} Raise: AuthException: raised if update operation fails """ response = self._http.post( MgmtV1.outbound_scim_update_path, - body=OutboundSCIMBase._compose_update_body(id, version, configuration, name), + body=OutboundSCIMBase._compose_update_body(app_id, version, configuration), ) return response.json() def delete_configuration( self, - id: str, + app_id: str, ): """ - Delete an existing outbound SCIM configuration. IMPORTANT: This action is - irreversible. Use carefully. + Delete the outbound SCIM configuration attached to the given federated SSO app. + IMPORTANT: This action is irreversible. Use carefully. Args: - id (str): The ID of the outbound SCIM configuration to delete. + app_id (str): The federated SSO application id. Raise: AuthException: raised if deletion operation fails """ - self._http.post(MgmtV1.outbound_scim_delete_path, body={"id": id}) + self._http.post(MgmtV1.outbound_scim_delete_path, body={"appId": app_id}) def load_configuration( self, - id: str, + app_id: str, ) -> dict: """ - Load an outbound SCIM configuration by ID. + Load the outbound SCIM configuration attached to the given federated SSO app. Args: - id (str): The ID of the outbound SCIM configuration to load. + app_id (str): The federated SSO application id. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , "appId": , - "configuration": {...}, "enabled": , - "lastExportTime": , "lastProcessingTime": , - "failures": , "version": }} + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} Raise: AuthException: raised if load operation fails """ - response = self._http.get(f"{MgmtV1.outbound_scim_load_path}/{id}") + response = self._http.get(f"{MgmtV1.outbound_scim_load_path}/{app_id}") return response.json() def set_enabled( self, - id: str, + app_id: str, enabled: bool, ) -> dict: """ - Enable or disable an outbound SCIM configuration. + Enable or disable the outbound SCIM configuration attached to the given + federated SSO app. Args: - id (str): The ID of the outbound SCIM configuration to update. + app_id (str): The federated SSO application id. enabled (bool): Whether the SCIM configuration should be enabled. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , "enabled": , ...}} + {"configuration": {"appId": , "enabled": , ...}} Raise: AuthException: raised if the operation fails """ response = self._http.post( MgmtV1.outbound_scim_set_enabled_path, - body={"id": id, "enabled": enabled}, + body={"appId": app_id, "enabled": enabled}, ) return response.json() diff --git a/descope/management/outbound_scim_async.py b/descope/management/outbound_scim_async.py index e08b82496..211bf19ba 100644 --- a/descope/management/outbound_scim_async.py +++ b/descope/management/outbound_scim_async.py @@ -12,125 +12,123 @@ class OutboundSCIMAsync(OutboundSCIMBase, AsyncHTTPBase): async def create_configuration( self, - name: str, app_id: str, configuration: Optional[dict] = None, ) -> dict: """ - Create a new outbound SCIM configuration. The configuration ID is provisioned - automatically by Descope and returned in the response. + Create a new outbound SCIM configuration on the federated SSO application + identified by app_id. The connector name is derived server-side from the app. Args: - name (str): The outbound SCIM configuration's name. - app_id (str): The ID of the outbound application this SCIM configuration binds to. + app_id (str): The federated SSO application id this SCIM configuration binds to. configuration (dict): Optional provider-specific SCIM configuration dictionary. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , "appId": , - "configuration": {...}, "enabled": , - "lastExportTime": , "lastProcessingTime": , - "failures": , "version": }} + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} Raise: AuthException: raised if create operation fails """ response = await self._http.post( MgmtV1.outbound_scim_create_path, - body=OutboundSCIMBase._compose_create_body(name, app_id, configuration), + body=OutboundSCIMBase._compose_create_body(app_id, configuration), ) return response.json() async def update_configuration( self, - id: str, + app_id: str, version: int, configuration: Optional[dict] = None, - name: Optional[str] = None, ) -> dict: """ - Update an existing outbound SCIM configuration. The version must match the - currently stored version — otherwise the update is rejected as a conflict. + Update the outbound SCIM configuration attached to the given federated SSO app. + The version must match the currently stored version — otherwise the update is + rejected as a conflict. Args: - id (str): The ID of the outbound SCIM configuration to update. + app_id (str): The federated SSO application id. version (int): The currently stored version, used for optimistic concurrency. configuration (dict): Optional updated provider-specific SCIM configuration. - name (str): Optional updated configuration name. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , ...}} + {"configuration": {"appId": , ...}} Raise: AuthException: raised if update operation fails """ response = await self._http.post( MgmtV1.outbound_scim_update_path, - body=OutboundSCIMBase._compose_update_body(id, version, configuration, name), + body=OutboundSCIMBase._compose_update_body(app_id, version, configuration), ) return response.json() async def delete_configuration( self, - id: str, + app_id: str, ): """ - Delete an existing outbound SCIM configuration. IMPORTANT: This action is - irreversible. Use carefully. + Delete the outbound SCIM configuration attached to the given federated SSO app. + IMPORTANT: This action is irreversible. Use carefully. Args: - id (str): The ID of the outbound SCIM configuration to delete. + app_id (str): The federated SSO application id. Raise: AuthException: raised if deletion operation fails """ - await self._http.post(MgmtV1.outbound_scim_delete_path, body={"id": id}) + await self._http.post(MgmtV1.outbound_scim_delete_path, body={"appId": app_id}) async def load_configuration( self, - id: str, + app_id: str, ) -> dict: """ - Load an outbound SCIM configuration by ID. + Load the outbound SCIM configuration attached to the given federated SSO app. Args: - id (str): The ID of the outbound SCIM configuration to load. + app_id (str): The federated SSO application id. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , "appId": , - "configuration": {...}, "enabled": , - "lastExportTime": , "lastProcessingTime": , - "failures": , "version": }} + {"configuration": {"appId": , "configuration": {...}, + "enabled": , "lastExportTime": , + "lastProcessingTime": , "failures": , + "version": }} Raise: AuthException: raised if load operation fails """ - response = await self._http.get(f"{MgmtV1.outbound_scim_load_path}/{id}") + response = await self._http.get(f"{MgmtV1.outbound_scim_load_path}/{app_id}") return response.json() async def set_enabled( self, - id: str, + app_id: str, enabled: bool, ) -> dict: """ - Enable or disable an outbound SCIM configuration. + Enable or disable the outbound SCIM configuration attached to the given + federated SSO app. Args: - id (str): The ID of the outbound SCIM configuration to update. + app_id (str): The federated SSO application id. enabled (bool): Whether the SCIM configuration should be enabled. Return value (dict): Return dict in the format - {"configuration": {"id": , "name": , "enabled": , ...}} + {"configuration": {"appId": , "enabled": , ...}} Raise: AuthException: raised if the operation fails """ response = await self._http.post( MgmtV1.outbound_scim_set_enabled_path, - body={"id": id, "enabled": enabled}, + body={"appId": app_id, "enabled": enabled}, ) return response.json() diff --git a/tests/management/test_outbound_scim.py b/tests/management/test_outbound_scim.py index 8eea3c563..d914d6173 100644 --- a/tests/management/test_outbound_scim.py +++ b/tests/management/test_outbound_scim.py @@ -9,8 +9,6 @@ CONFIG_RESPONSE = { "configuration": { - "id": "scim1", - "name": "Test SCIM", "appId": "app1", "configuration": {"baseUrl": "https://scim.example.com"}, "enabled": True, @@ -35,7 +33,6 @@ async def test_create_configuration_success(self, client_factory): with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: response = await client.invoke( client.mgmt.outbound_scim.create_configuration( - "Test SCIM", "app1", {"baseUrl": "https://scim.example.com"}, ) @@ -48,7 +45,6 @@ async def test_create_configuration_success(self, client_factory): headers=MGMT_HEADERS, params=None, json={ - "name": "Test SCIM", "appId": "app1", "configuration": {"baseUrl": "https://scim.example.com"}, }, @@ -59,7 +55,7 @@ async def test_create_configuration_minimal_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: - response = await client.invoke(client.mgmt.outbound_scim.create_configuration("Test SCIM", "app1")) + response = await client.invoke(client.mgmt.outbound_scim.create_configuration("app1")) assert response == CONFIG_RESPONSE assert_http_called( mock_post, @@ -67,7 +63,7 @@ async def test_create_configuration_minimal_success(self, client_factory): f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_create_path}", headers=MGMT_HEADERS, params=None, - json={"name": "Test SCIM", "appId": "app1"}, + json={"appId": "app1"}, follow_redirects=False, ) @@ -76,7 +72,7 @@ async def test_create_configuration_failure(self, client_factory): with client.mock_mgmt_post(make_response(status=500)): with pytest.raises(AuthException): - await client.invoke(client.mgmt.outbound_scim.create_configuration("Test SCIM", "app1")) + await client.invoke(client.mgmt.outbound_scim.create_configuration("app1")) async def test_update_configuration_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") @@ -84,10 +80,9 @@ async def test_update_configuration_success(self, client_factory): with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: response = await client.invoke( client.mgmt.outbound_scim.update_configuration( - "scim1", + "app1", 3, {"baseUrl": "https://scim.example.com"}, - "Updated Name", ) ) assert response == CONFIG_RESPONSE @@ -98,32 +93,25 @@ async def test_update_configuration_success(self, client_factory): headers=MGMT_HEADERS, params=None, json={ - "id": "scim1", + "appId": "app1", "version": 3, - "name": "Updated Name", "configuration": {"baseUrl": "https://scim.example.com"}, }, follow_redirects=False, ) - async def test_update_configuration_without_name_success(self, client_factory): + async def test_update_configuration_no_config_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: - await client.invoke( - client.mgmt.outbound_scim.update_configuration("scim1", 3, {"baseUrl": "https://scim.example.com"}) - ) + await client.invoke(client.mgmt.outbound_scim.update_configuration("app1", 3)) assert_http_called( mock_post, client.mode, f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_update_path}", headers=MGMT_HEADERS, params=None, - json={ - "id": "scim1", - "version": 3, - "configuration": {"baseUrl": "https://scim.example.com"}, - }, + json={"appId": "app1", "version": 3}, follow_redirects=False, ) @@ -132,20 +120,20 @@ async def test_update_configuration_failure(self, client_factory): with client.mock_mgmt_post(make_response(status=500)): with pytest.raises(AuthException): - await client.invoke(client.mgmt.outbound_scim.update_configuration("scim1", 3, {})) + await client.invoke(client.mgmt.outbound_scim.update_configuration("app1", 3, {})) async def test_delete_configuration_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") with client.mock_mgmt_post(make_response({})) as mock_post: - await client.invoke(client.mgmt.outbound_scim.delete_configuration("scim1")) + await client.invoke(client.mgmt.outbound_scim.delete_configuration("app1")) assert_http_called( mock_post, client.mode, f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_delete_path}", headers=MGMT_HEADERS, params=None, - json={"id": "scim1"}, + json={"appId": "app1"}, follow_redirects=False, ) @@ -154,18 +142,18 @@ async def test_delete_configuration_failure(self, client_factory): with client.mock_mgmt_post(make_response(status=500)): with pytest.raises(AuthException): - await client.invoke(client.mgmt.outbound_scim.delete_configuration("scim1")) + await client.invoke(client.mgmt.outbound_scim.delete_configuration("app1")) async def test_load_configuration_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") with client.mock_mgmt_get(make_response(CONFIG_RESPONSE)) as mock_get: - response = await client.invoke(client.mgmt.outbound_scim.load_configuration("scim1")) + response = await client.invoke(client.mgmt.outbound_scim.load_configuration("app1")) assert response == CONFIG_RESPONSE assert_http_called( mock_get, client.mode, - f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_load_path}/scim1", + f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_load_path}/app1", headers=MGMT_HEADERS, params=None, follow_redirects=True, @@ -176,13 +164,13 @@ async def test_load_configuration_failure(self, client_factory): with client.mock_mgmt_get(make_response(status=500)): with pytest.raises(AuthException): - await client.invoke(client.mgmt.outbound_scim.load_configuration("scim1")) + await client.invoke(client.mgmt.outbound_scim.load_configuration("app1")) async def test_set_enabled_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: - response = await client.invoke(client.mgmt.outbound_scim.set_enabled("scim1", True)) + response = await client.invoke(client.mgmt.outbound_scim.set_enabled("app1", True)) assert response == CONFIG_RESPONSE assert_http_called( mock_post, @@ -190,7 +178,7 @@ async def test_set_enabled_success(self, client_factory): f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_set_enabled_path}", headers=MGMT_HEADERS, params=None, - json={"id": "scim1", "enabled": True}, + json={"appId": "app1", "enabled": True}, follow_redirects=False, ) @@ -198,14 +186,14 @@ async def test_set_enabled_false_success(self, client_factory): client = client_factory.make(PROJECT_ID, PUBLIC_KEY_DICT, False, "key") with client.mock_mgmt_post(make_response(CONFIG_RESPONSE)) as mock_post: - await client.invoke(client.mgmt.outbound_scim.set_enabled("scim1", False)) + await client.invoke(client.mgmt.outbound_scim.set_enabled("app1", False)) assert_http_called( mock_post, client.mode, f"{DEFAULT_BASE_URL}{MgmtV1.outbound_scim_set_enabled_path}", headers=MGMT_HEADERS, params=None, - json={"id": "scim1", "enabled": False}, + json={"appId": "app1", "enabled": False}, follow_redirects=False, ) @@ -214,34 +202,31 @@ async def test_set_enabled_failure(self, client_factory): with client.mock_mgmt_post(make_response(status=500)): with pytest.raises(AuthException): - await client.invoke(client.mgmt.outbound_scim.set_enabled("scim1", True)) + await client.invoke(client.mgmt.outbound_scim.set_enabled("app1", True)) def test_compose_create_body(self): - body = OutboundSCIM._compose_create_body("Test SCIM", "app1", {"baseUrl": "https://scim.example.com"}) + body = OutboundSCIM._compose_create_body("app1", {"baseUrl": "https://scim.example.com"}) assert body == { - "name": "Test SCIM", "appId": "app1", "configuration": {"baseUrl": "https://scim.example.com"}, } def test_compose_create_body_without_configuration(self): - body = OutboundSCIM._compose_create_body("Test SCIM", "app1") - assert body == {"name": "Test SCIM", "appId": "app1"} + body = OutboundSCIM._compose_create_body("app1") + assert body == {"appId": "app1"} def test_compose_update_body(self): body = OutboundSCIM._compose_update_body( - "scim1", + "app1", 5, {"baseUrl": "https://scim.example.com"}, - "New Name", ) assert body == { - "id": "scim1", + "appId": "app1", "version": 5, - "name": "New Name", "configuration": {"baseUrl": "https://scim.example.com"}, } - def test_compose_update_body_without_optionals(self): - body = OutboundSCIM._compose_update_body("scim1", 5) - assert body == {"id": "scim1", "version": 5} + def test_compose_update_body_without_configuration(self): + body = OutboundSCIM._compose_update_body("app1", 5) + assert body == {"appId": "app1", "version": 5}