Add tenant role ID and name parameters for user search (#612)

Author: talberry

descope/management/common.py

@@ -364,3 +364,8 @@ def sort_to_dict(sort: List[Sort]) -> list:
                 }
             )
     return sort_list
+
+def map_to_values_object(input_map: dict):
+    if not input_map:
+        return {}
+    return {k: {"values": v} for k, v in input_map.items()}

descope/management/user.py

@@ -10,6 +10,7 @@
     Sort,
     associated_tenants_to_dict,
     sort_to_dict,
+    map_to_values_object,
 )
 from descope.management.user_pwd import UserPassword
 
@@ -638,6 +639,8 @@ def search_all(
         from_modified_time: Optional[int] = None,
         to_modified_time: Optional[int] = None,
         user_ids: Optional[List[str]] = None,
+        tenant_role_ids: Optional[dict] = None,
+        tenant_role_names: Optional[dict] = None,
     ) -> dict:
         """
         Search all users.
@@ -662,6 +665,8 @@ def search_all(
         from_modified_time (int): Optional int, only include users whose last modification/update occurred on or after this time (in Unix epoch milliseconds)
         to_modified_time (int): Optional int, only include users whose last modification/update occurred on or before this time (in Unix epoch milliseconds)
         user_ids (List[str]): Optional list of user IDs to filter by
+        tenant_role_ids (dict): Optional mapping of tenant ID to list of role IDs.
+        tenant_role_names (dict): Optional mapping of tenant ID to list of role names.
 
         Return value (dict):
         Return dict in the format
@@ -726,7 +731,12 @@ def search_all(
             body["fromModifiedTime"] = from_modified_time
         if to_modified_time is not None:
             body["toModifiedTime"] = to_modified_time
-
+            
+        if tenant_role_ids is not None:
+            body["tenantRoleIds"] = map_to_values_object(tenant_role_ids)
+        if tenant_role_names is not None:
+            body["tenantRoleNames"] = map_to_values_object(tenant_role_names)
+        
         response = self._auth.do_post(
             MgmtV1.users_search_path,
             body=body,
@@ -752,6 +762,8 @@ def search_all_test_users(
         to_created_time: Optional[int] = None,
         from_modified_time: Optional[int] = None,
         to_modified_time: Optional[int] = None,
+        tenant_role_ids: Optional[dict] = None,
+        tenant_role_names: Optional[dict] = None,
     ) -> dict:
         """
         Search all test users.
@@ -773,6 +785,8 @@ def search_all_test_users(
         to_created_time (int): Optional int, only include users who were created on or before this time (in Unix epoch milliseconds)
         from_modified_time (int): Optional int, only include users whose last modification/update occurred on or after this time (in Unix epoch milliseconds)
         to_modified_time (int): Optional int, only include users whose last modification/update occurred on or before this time (in Unix epoch milliseconds)
+        tenant_role_ids (dict): Optional mapping of tenant ID to list of role IDs.
+        tenant_role_names (dict): Optional mapping of tenant ID to list of role names.
 
         Return value (dict):
         Return dict in the format
@@ -834,6 +848,11 @@ def search_all_test_users(
             body["fromModifiedTime"] = from_modified_time
         if to_modified_time is not None:
             body["toModifiedTime"] = to_modified_time
+            
+        if tenant_role_ids is not None:
+            body["tenantRoleIds"] = map_to_values_object(tenant_role_ids)
+        if tenant_role_names is not None:
+            body["tenantRoleNames"] = map_to_values_object(tenant_role_names)
 
         response = self._auth.do_post(
             MgmtV1.test_users_search_path,

tests/management/test_user.py

@@ -1047,6 +1047,45 @@ def test_search_all(self):
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
 
+        # Test success flow with tenant_role_ids and tenant_role_names
+        with patch("requests.post") as mock_post:
+            network_resp = mock.Mock()
+            network_resp.ok = True
+            network_resp.json.return_value = json.loads(
+                """{"users": [{"id": "u1"}, {"id": "u2"}]}"""
+            )
+            mock_post.return_value = network_resp
+            resp = self.client.mgmt.user.search_all(
+                tenant_role_ids={"tenant1": ["roleA", "roleB"]},
+                tenant_role_names={"tenant2": ["admin", "user"]},
+            )
+            users = resp["users"]
+            self.assertEqual(len(users), 2)
+            self.assertEqual(users[0]["id"], "u1")
+            self.assertEqual(users[1]["id"], "u2")
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.users_search_path}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={
+                    "tenantIds": [],
+                    "roleNames": [],
+                    "limit": 0,
+                    "page": 0,
+                    "testUsersOnly": False,
+                    "withTestUser": False,
+                    "tenantRoleIds": {"tenant1": {"values": ["roleA", "roleB"]}},
+                    "tenantRoleNames": {"tenant2": {"values": ["admin", "user"]}},
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
     def test_search_all_test_users(self):
         # Test failed flows
         with patch("requests.post") as mock_post:
@@ -1254,6 +1293,45 @@ def test_search_all_test_users(self):
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
 
+        # Test success flow with tenant_role_ids and tenant_role_names
+        with patch("requests.post") as mock_post:
+            network_resp = mock.Mock()
+            network_resp.ok = True
+            network_resp.json.return_value = json.loads(
+                """{"users": [{"id": "u1"}, {"id": "u2"}]}"""
+            )
+            mock_post.return_value = network_resp
+            resp = self.client.mgmt.user.search_all_test_users(
+                tenant_role_ids={"tenant1": ["roleA", "roleB"]},
+                tenant_role_names={"tenant2": ["admin", "user"]},
+            )
+            users = resp["users"]
+            self.assertEqual(len(users), 2)
+            self.assertEqual(users[0]["id"], "u1")
+            self.assertEqual(users[1]["id"], "u2")
+            mock_post.assert_called_with(
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.test_users_search_path}",
+                headers={
+                    **common.default_headers,
+                    "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
+                    "x-descope-project-id": self.dummy_project_id,
+                },
+                params=None,
+                json={
+                    "tenantIds": [],
+                    "roleNames": [],
+                    "limit": 0,
+                    "page": 0,
+                    "testUsersOnly": True,
+                    "withTestUser": True,
+                    "tenantRoleIds": {"tenant1": {"values": ["roleA", "roleB"]}},
+                    "tenantRoleNames": {"tenant2": {"values": ["admin", "user"]}},
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
     def test_get_provider_token(self):
         # Test failed flows
         with patch("requests.get") as mock_post: