Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -680,7 +680,7 @@ descope_client.mgmt.user.logout_user("<login-id>")
descope_client.mgmt.user.logout_user_by_user_id("<user-id>")

# Search all users, optionally according to tenant and/or role filter
# results can be paginated using the limit and page parameters
# results can be paginated using the limit and page parameters, as well as by time with the from_created_time, to_created_time, from_modified_time, and to_modified_time
users_resp = descope_client.mgmt.user.search_all(tenant_ids=["my-tenant-id"])
users = users_resp["users"]
for user in users:
Expand Down
34 changes: 34 additions & 0 deletions descope/management/user.py
Original file line number Diff line number Diff line change
Expand Up @@ -595,6 +595,10 @@ def search_all(
sort: Optional[List[Sort]] = None,
text: Optional[str] = None,
login_ids: Optional[List[str]] = None,
from_created_time: Optional[int] = None,
to_created_time: Optional[int] = None,
from_modified_time: Optional[int] = None,
to_modified_time: Optional[int] = None,
) -> dict:
"""
Search all users.
Expand All @@ -614,6 +618,10 @@ def search_all(
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.
from_created_time (int): Optional int, only include users who were created on or after this time (in Unix epoch milliseconds)
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)

Return value (dict):
Return dict in the format
Expand Down Expand Up @@ -667,6 +675,15 @@ def search_all(
if sort is not None:
body["sort"] = sort_to_dict(sort)

if from_created_time is not None:
body["fromCreatedTime"] = from_created_time
if to_created_time is not None:
body["toCreatedTime"] = to_created_time
if from_modified_time is not None:
body["fromModifiedTime"] = from_modified_time
if to_modified_time is not None:
body["toModifiedTime"] = to_modified_time

response = self._auth.do_post(
MgmtV1.users_search_path,
body=body,
Expand All @@ -688,6 +705,10 @@ def search_all_test_users(
sort: Optional[List[Sort]] = None,
text: Optional[str] = None,
login_ids: Optional[List[str]] = None,
from_created_time: Optional[int] = None,
to_created_time: Optional[int] = None,
from_modified_time: Optional[int] = None,
to_modified_time: Optional[int] = None,
) -> dict:
"""
Search all test users.
Expand All @@ -705,6 +726,10 @@ def search_all_test_users(
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.
from_created_time (int): Optional int, only include users who were created on or after this time (in Unix epoch milliseconds)
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)

Return value (dict):
Return dict in the format
Expand Down Expand Up @@ -758,6 +783,15 @@ def search_all_test_users(
if sort is not None:
body["sort"] = sort_to_dict(sort)

if from_created_time is not None:
body["fromCreatedTime"] = from_created_time
if to_created_time is not None:
body["toCreatedTime"] = to_created_time
if from_modified_time is not None:
body["fromModifiedTime"] = from_modified_time
if to_modified_time is not None:
body["toModifiedTime"] = to_modified_time

response = self._auth.do_post(
MgmtV1.test_users_search_path,
body=body,
Expand Down
87 changes: 87 additions & 0 deletions tests/management/test_user.py
Original file line number Diff line number Diff line change
Expand Up @@ -965,6 +965,51 @@ def test_search_all(self):
verify=True,
timeout=DEFAULT_TIMEOUT_SECONDS,
)

# Test success flow with time parameters
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(
from_created_time=100,
to_created_time=200,
from_modified_time=300,
to_modified_time=400,
limit=10,
page=0,
)
users = resp["users"]
self.assertEqual(len(users), 2)
self.assertEqual(users[0]["id"], "u1")
self.assertEqual(users[1]["id"], "u2")
# Verify the request body includes our time parameters
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}",
},
params=None,
json={
"tenantIds": [],
"roleNames": [],
"limit": 10,
"page": 0,
"testUsersOnly": False,
"withTestUser": False,
"fromCreatedTime": 100,
"toCreatedTime": 200,
"fromModifiedTime": 300,
"toModifiedTime": 400,
},
allow_redirects=False,
verify=True,
timeout=DEFAULT_TIMEOUT_SECONDS,
)

def test_search_all_test_users(self):
# Test failed flows
Expand Down Expand Up @@ -1127,6 +1172,48 @@ def test_search_all_test_users(self):
timeout=DEFAULT_TIMEOUT_SECONDS,
)

# Test success flow with time parameters
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"}]}""")
mock_post.return_value = network_resp
resp = self.client.mgmt.user.search_all_test_users(
from_created_time=100,
to_created_time=200,
from_modified_time=300,
to_modified_time=400,
limit=10,
page=0,
)
users = resp["users"]
self.assertEqual(len(users), 1)
self.assertEqual(users[0]["id"], "u1")
# Verify the request body includes our time parameters
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": [],
"roleNames": [],
"limit": 10,
"page": 0,
"testUsersOnly": True,
"withTestUser": True,
"fromCreatedTime": 100,
"toCreatedTime": 200,
"fromModifiedTime": 300,
"toModifiedTime": 400,
},
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:
Expand Down
Loading