diff --git a/README.md b/README.md index 2f774868a..6c9ba9fd9 100644 --- a/README.md +++ b/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. diff --git a/descope/management/common.py b/descope/management/common.py index f7be476ea..331529063 100644 --- a/descope/management/common.py +++ b/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" diff --git a/descope/management/user.py b/descope/management/user.py index 765b58e1c..917a4ff54 100644 --- a/descope/management/user.py +++ b/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, diff --git a/tests/management/test_user.py b/tests/management/test_user.py index 3e3612ba5..bd9e6be95 100644 --- a/tests/management/test_user.py +++ b/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=["a@b.com"], + ) + 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": ["a@b.com"], + "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: