Add test user apis (#454)

* Add test user apis

* fix test

Author: guyp-descope

README.md

@@ -1393,6 +1393,13 @@ descope_client.mgmt.user.create_test_user(
     ],
 )
 
+# Search all test users, optionally according to tenant and/or role filter
+# results can be paginated using the limit and page parameters
+users_resp = descope_client.mgmt.user.search_all_test_users()
+users = users_resp["users"]
+    for user in users:
+        # Do something
+
 # Now test user got created, and this user will be available until you delete it,
 # you can use any management operation for test user CRUD.
 # You can also delete all test users.

descope/management/common.py

@@ -21,6 +21,7 @@ class MgmtV1:
 
     # user
     user_create_path = "/v1/mgmt/user/create"
+    test_user_create_path = "/v1/mgmt/user/create/test"
     user_create_batch_path = "/v1/mgmt/user/create/batch"
     user_update_path = "/v1/mgmt/user/update"
     user_patch_path = "/v1/mgmt/user/patch"
@@ -29,6 +30,7 @@ class MgmtV1:
     user_delete_all_test_users_path = "/v1/mgmt/user/test/delete/all"
     user_load_path = "/v1/mgmt/user"
     users_search_path = "/v2/mgmt/user/search"
+    test_users_search_path = "/v2/mgmt/user/search/test"
     user_get_provider_token = "/v1/mgmt/user/provider/token"
     user_update_status_path = "/v1/mgmt/user/update/status"
     user_update_login_id_path = "/v1/mgmt/user/update/loginid"

descope/management/user.py

@@ -178,7 +178,7 @@ def create_test_user(
         user_tenants = [] if user_tenants is None else user_tenants
 
         response = self._auth.do_post(
-            MgmtV1.user_create_path,
+            MgmtV1.test_user_create_path,
             User._compose_create_body(
                 login_id,
                 email,
@@ -199,6 +199,7 @@ def create_test_user(
                 None,
                 None,
                 additional_login_ids,
+                sso_app_ids,
             ),
             pswd=self._auth.management_key,
         )
@@ -671,6 +672,97 @@ def search_all(
         )
         return response.json()
 
+    def search_all_test_users(
+        self,
+        tenant_ids: Optional[List[str]] = None,
+        role_names: Optional[List[str]] = None,
+        limit: int = 0,
+        page: int = 0,
+        custom_attributes: Optional[dict] = None,
+        statuses: Optional[List[str]] = None,
+        emails: Optional[List[str]] = None,
+        phones: Optional[List[str]] = None,
+        sso_app_ids: Optional[List[str]] = None,
+        sort: Optional[List[Sort]] = None,
+        text: Optional[str] = None,
+        login_ids: Optional[List[str]] = None,
+    ) -> dict:
+        """
+        Search all test users.
+
+        Args:
+        tenant_ids (List[str]): Optional list of tenant IDs to filter by
+        role_names (List[str]): Optional list of role names to filter by
+        limit (int): Optional limit of the number of users returned. Leave empty for default.
+        page (int): Optional pagination control. Pages start at 0 and must be non-negative.
+        custom_attributes (dict): Optional search for a attribute with a given value
+        statuses (List[str]): Optional list of statuses to search for ("enabled", "disabled", "invited")
+        emails (List[str]): Optional list of emails to search for
+        phones (List[str]): Optional list of phones to search for
+        sso_app_ids (List[str]): Optional list of SSO application IDs to filter by
+        text (str): Optional string, allows free text search among all user's attributes.
+        login_ids (List[str]): Optional list of login ids
+        sort (List[Sort]): Optional List[dict], allows to sort by fields.
+
+        Return value (dict):
+        Return dict in the format
+             {"users": []}
+        "users" contains a list of all of the found users and their information
+
+        Raise:
+        AuthException: raised if search operation fails
+        """
+        tenant_ids = [] if tenant_ids is None else tenant_ids
+        role_names = [] if role_names is None else role_names
+
+        if limit < 0:
+            raise AuthException(
+                400, ERROR_TYPE_INVALID_ARGUMENT, "limit must be non-negative"
+            )
+
+        if page < 0:
+            raise AuthException(
+                400, ERROR_TYPE_INVALID_ARGUMENT, "page must be non-negative"
+            )
+        body = {
+            "tenantIds": tenant_ids,
+            "roleNames": role_names,
+            "limit": limit,
+            "page": page,
+            "testUsersOnly": True,
+            "withTestUser": True,
+        }
+        if statuses is not None:
+            body["statuses"] = statuses
+
+        if emails is not None:
+            body["emails"] = emails
+
+        if phones is not None:
+            body["phones"] = phones
+
+        if custom_attributes is not None:
+            body["customAttributes"] = custom_attributes
+
+        if sso_app_ids is not None:
+            body["ssoAppIds"] = sso_app_ids
+
+        if login_ids is not None:
+            body["loginIds"] = login_ids
+
+        if text is not None:
+            body["text"] = text
+
+        if sort is not None:
+            body["sort"] = sort_to_dict(sort)
+
+        response = self._auth.do_post(
+            MgmtV1.test_users_search_path,
+            body=body,
+            pswd=self._auth.management_key,
+        )
+        return response.json()
+
     def get_provider_token(
         self,
         login_id: str,

tests/management/test_user.py

@@ -180,7 +180,7 @@ def test_create_test_user(self):
             user = resp["user"]
             self.assertEqual(user["id"], "u1")
             mock_post.assert_called_with(
-                f"{common.DEFAULT_BASE_URL}{MgmtV1.user_create_path}",
+                f"{common.DEFAULT_BASE_URL}{MgmtV1.test_user_create_path}",
                 headers={
                     **common.default_headers,
                     "Authorization": f"Bearer {self.dummy_project_id}:{self.dummy_management_key}",
@@ -964,6 +964,167 @@ def test_search_all(self):
                 timeout=DEFAULT_TIMEOUT_SECONDS,
             )
 
+    def test_search_all_test_users(self):
+        # Test failed flows
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = False
+            self.assertRaises(
+                AuthException,
+                self.client.mgmt.user.search_all_test_users,
+                ["t1, t2"],
+                ["r1", "r2"],
+            )
+
+        with patch("requests.post") as mock_post:
+            mock_post.return_value.ok = True
+            self.assertRaises(
+                AuthException,
+                self.client.mgmt.user.search_all_test_users,
+                [],
+                [],
+                -1,
+                0,
+            )
+
+            self.assertRaises(
+                AuthException,
+                self.client.mgmt.user.search_all_test_users,
+                [],
+                [],
+                0,
+                -1,
+            )
+
+        # Test success flow
+        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(
+                ["t1, t2"],
+                ["r1", "r2"],
+                sso_app_ids=["app1"],
+                login_ids=["l1"],
+            )
+            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}",
+                },
+                params=None,
+                json={
+                    "tenantIds": ["t1, t2"],
+                    "roleNames": ["r1", "r2"],
+                    "limit": 0,
+                    "page": 0,
+                    "testUsersOnly": True,
+                    "withTestUser": True,
+                    "ssoAppIds": ["app1"],
+                    "loginIds": ["l1"],
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+        # Test success flow with text and sort
+        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
+            sort = [Sort(field="kuku", desc=True), Sort(field="bubu")]
+            resp = self.client.mgmt.user.search_all_test_users(
+                ["t1, t2"],
+                ["r1", "r2"],
+                sso_app_ids=["app1"],
+                text="blue",
+                sort=sort,
+            )
+            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}",
+                },
+                params=None,
+                json={
+                    "tenantIds": ["t1, t2"],
+                    "roleNames": ["r1", "r2"],
+                    "limit": 0,
+                    "page": 0,
+                    "testUsersOnly": True,
+                    "withTestUser": True,
+                    "ssoAppIds": ["app1"],
+                    "text": "blue",
+                    "sort": [
+                        {"desc": True, "field": "kuku"},
+                        {"desc": False, "field": "bubu"},
+                    ],
+                },
+                allow_redirects=False,
+                verify=True,
+                timeout=DEFAULT_TIMEOUT_SECONDS,
+            )
+
+        # Test success flow with custom attributes
+        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(
+                ["t1, t2"],
+                ["r1", "r2"],
+                custom_attributes={"ak": "av"},
+                statuses=["invited"],
+                phones=["+111111"],
+                emails=["[email protected]"],
+            )
+            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}",
+                },
+                params=None,
+                json={
+                    "tenantIds": ["t1, t2"],
+                    "roleNames": ["r1", "r2"],
+                    "limit": 0,
+                    "page": 0,
+                    "testUsersOnly": True,
+                    "withTestUser": True,
+                    "customAttributes": {"ak": "av"},
+                    "statuses": ["invited"],
+                    "emails": ["[email protected]"],
+                    "phones": ["+111111"],
+                },
+                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: