gmakstutis/ragflow
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

[OND211-2329]: Updated user list pagination response.

Browse source
This commit is contained in:
Hetavi Shah 2025-12-01 17:08:00 +05:30
parent 9f4e0f48ed
commit d247aabfae
2 changed files with 107 additions and 70 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

97
api/apps/user_app.py
View file

@ -1260,25 +1260,28 @@ def list_users() -> Response:
type: object
properties:
data:
type: array
items:
type: object
properties:
id:
type: string
description: User ID.
email:
type: string
description: User email.
nickname:
type: string
description: User nickname.
is_superuser:
type: boolean
description: Whether the user is a superuser.
total:
type: integer
description: Total number of users.
type: object
properties:
users:
type: array
items:
type: object
properties:
id:
type: string
description: User ID.
email:
type: string
description: User email.
nickname:
type: string
description: User nickname.
is_superuser:
type: boolean
description: Whether the user is a superuser.
total:
type: integer
description: Total number of users.
401:
description: Unauthorized - authentication required.
schema:
@ -1327,28 +1330,7 @@ def list_users() -> Response:
email_filter = request.args.get("email")
# Query users
if email_filter:
# Validate email format if provided (allow + in local part)
email_match: Optional[Match[str]] = re.match(
r"^[\w\._\+-]+@([\w_-]+\.)+[\w-]{2,}$", email_filter
)
if not email_match:
return get_json_result(
data=False,
message=f"Invalid email address: {email_filter}!",
code=RetCode.OPERATING_ERROR,
)
users_query = UserService.query(email=email_filter)
users_list: List[User] = list(users_query)
else:
users_list: List[User] = UserService.get_all_users()
# Convert users to dictionaries
users_data: List[Dict[str, Any]] = [
user.to_dict() for user in users_list
]
# Validate pagination parameters if provided
if page is not None and page_size is not None:
if page < 1:
return get_json_result(
@ -1363,12 +1345,37 @@ def list_users() -> Response:
code=RetCode.ARGUMENT_ERROR,
)
start_idx: int = (page - 1) * page_size
end_idx: int = start_idx + page_size
users_data = users_data[start_idx:end_idx]
# Build query
if email_filter:
# Validate email format if provided (allow + in local part)
email_match: Optional[Match[str]] = re.match(
r"^[\w\._\+-]+@([\w_-]+\.)+[\w-]{2,}$", email_filter
)
if not email_match:
return get_json_result(
data=False,
message=f"Invalid email address: {email_filter}!",
code=RetCode.OPERATING_ERROR,
)
users_query = UserService.model.select().where(UserService.model.email == email_filter)
else:
users_query = UserService.model.select()
# Get total count before pagination
total: int = users_query.count()
# Apply pagination at database level
if page is not None and page_size is not None:
users_query = users_query.paginate(page, page_size)
# Convert to list and dictionaries
users_list: List[User] = list(users_query)
users_data: List[Dict[str, Any]] = [
user.to_dict() for user in users_list
]
return get_json_result(
data=users_data,
data={"users": users_data, "total": total},
message=f"Retrieved {len(users_data)} user(s) successfully!",
)
except Exception as e:

80
test/testcases/test_http_api/test_user_management/test_list_user.py
View file

@ -102,10 +102,13 @@ class TestUserList:
list_res: dict[str, Any] = list_users(web_api_auth)
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert len(list_res["data"]) >= 1
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
assert isinstance(list_res["data"]["users"], list)
assert len(list_res["data"]["users"]) >= 1
# Verify the created user is in the list
user_emails: list[str] = [u["email"] for u in list_res["data"]]
user_emails: list[str] = [u["email"] for u in list_res["data"]["users"]]
assert unique_email in user_emails
@pytest.mark.p1
@ -135,10 +138,13 @@ class TestUserList:
list_res: dict[str, Any] = list_users(web_api_auth)
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert len(list_res["data"]) >= len(created_emails)
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
assert isinstance(list_res["data"]["users"], list)
assert len(list_res["data"]["users"]) >= len(created_emails)
# Verify all created users are in the list
user_emails: list[str] = [u["email"] for u in list_res["data"]]
user_emails: list[str] = [u["email"] for u in list_res["data"]["users"]]
for email in created_emails:
assert email in user_emails
@ -166,10 +172,13 @@ class TestUserList:
params: dict[str, str] = {"email": unique_email}
list_res: dict[str, Any] = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert len(list_res["data"]) >= 1
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
assert isinstance(list_res["data"]["users"], list)
assert len(list_res["data"]["users"]) >= 1
# Verify all returned users have the filtered email
for user in list_res["data"]:
for user in list_res["data"]["users"]:
assert user["email"] == unique_email
@pytest.mark.p1
@ -191,8 +200,12 @@ class TestUserList:
params: dict[str, str] = {"email": nonexistent_email}
list_res: dict[str, Any] = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert len(list_res["data"]) == 0
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
assert isinstance(list_res["data"]["users"], list)
assert len(list_res["data"]["users"]) == 0
assert list_res["data"]["total"] == 0
@pytest.mark.p1
@pytest.mark.parametrize(
@ -240,9 +253,12 @@ class TestUserList:
if expected_valid:
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
assert isinstance(list_res["data"]["users"], list)
# Verify pagination limits
assert len(list_res["data"]) <= page_size
assert len(list_res["data"]["users"]) <= page_size
else:
assert list_res["code"] != 0
assert "must be greater than 0" in list_res["message"]
@ -276,19 +292,21 @@ class TestUserList:
# Get total count of all users to calculate pagination boundaries
list_res_all: dict[str, Any] = list_users(web_api_auth)
total_users: int = len(list_res_all["data"])
total_users: int = list_res_all["data"]["total"]
# Test first page
params: dict[str, int] = {"page": 1, "page_size": 2}
list_res: dict[str, Any] = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert len(list_res["data"]) == 2
assert len(list_res["data"]["users"]) == 2
assert list_res["data"]["total"] == total_users
# Test that pagination returns consistent page sizes
params = {"page": 2, "page_size": 2}
list_res = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert len(list_res["data"]) == 2
assert len(list_res["data"]["users"]) == 2
assert list_res["data"]["total"] == total_users
# Test last page (might have fewer items)
# Calculate expected last page: ceil(total_users / page_size)
@ -298,14 +316,16 @@ class TestUserList:
params = {"page": last_page, "page_size": page_size}
list_res = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert len(list_res["data"]) <= page_size
assert len(list_res["data"]["users"]) <= page_size
assert list_res["data"]["total"] == total_users
# Test page beyond available data
# Use a page number that's definitely beyond available data
params = {"page": total_users + 10, "page_size": 2}
list_res = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert len(list_res["data"]) == 0
assert len(list_res["data"]["users"]) == 0
assert list_res["data"]["total"] == total_users
@pytest.mark.p1
def test_list_users_response_structure(
@ -315,10 +335,14 @@ class TestUserList:
res: dict[str, Any] = list_users(web_api_auth)
assert res["code"] == 0
assert "data" in res
assert isinstance(res["data"], list)
assert isinstance(res["data"], dict)
assert "users" in res["data"]
assert "total" in res["data"]
assert isinstance(res["data"]["users"], list)
assert isinstance(res["data"]["total"], int)
if len(res["data"]) > 0:
user: dict[str, Any] = res["data"][0]
if len(res["data"]["users"]) > 0:
user: dict[str, Any] = res["data"]["users"][0]
# Verify user structure
assert "id" in user
assert "email" in user
@ -372,9 +396,12 @@ class TestUserList:
}
list_res: dict[str, Any] = list_users(web_api_auth, params=params)
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
# Should return at least the created user
assert len(list_res["data"]) >= 1
assert len(list_res["data"]["users"]) >= 1
assert list_res["data"]["total"] >= 1
@pytest.mark.p2
def test_list_users_performance_with_many_users(
@ -405,7 +432,10 @@ class TestUserList:
# List all users
list_res: dict[str, Any] = list_users(web_api_auth)
assert list_res["code"] == 0, list_res
assert isinstance(list_res["data"], list)
assert isinstance(list_res["data"], dict)
assert "users" in list_res["data"]
assert "total" in list_res["data"]
# Should return at least the created users
assert len(list_res["data"]) >= created_count
assert len(list_res["data"]["users"]) >= created_count
assert list_res["data"]["total"] >= created_count