feat: add system quality improvement plan and implementation
This commit is contained in:
@@ -114,7 +114,7 @@ class TestUser:
|
||||
|
||||
response = await user_api.logical_delete_user(user_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.status_code == 204
|
||||
|
||||
get_response = await user_api.get_user_by_id(user_id)
|
||||
assert get_response.status_code == 404
|
||||
@@ -137,7 +137,7 @@ class TestUser:
|
||||
|
||||
response = await user_api.restore_user(user_id)
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.status_code == 204
|
||||
|
||||
get_response = await user_api.get_user_by_id(user_id)
|
||||
assert get_response.status_code == 200
|
||||
@@ -191,3 +191,150 @@ class TestUser:
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_users_by_page_success(self, authenticated_client, test_user_data, cleanup_user):
|
||||
"""测试分页获取用户成功"""
|
||||
import time
|
||||
user_api = UserAPI(authenticated_client)
|
||||
|
||||
timestamp = int(time.time() * 1000)
|
||||
for i in range(5):
|
||||
user_data = test_user_data.copy()
|
||||
user_data["username"] = f"testuser_{timestamp}_{i}"
|
||||
user_data["email"] = f"testuser_{timestamp}_{i}@example.com"
|
||||
create_response = await user_api.create_user(user_data)
|
||||
cleanup_user.append(create_response.json()["id"])
|
||||
|
||||
response = await user_api.get_users_by_page(page=0, size=10)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "content" in data
|
||||
assert "totalElements" in data
|
||||
assert "totalPages" in data
|
||||
assert "currentPage" in data
|
||||
assert "pageSize" in data
|
||||
assert len(data["content"]) <= 10
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_users_by_page_with_sort(self, authenticated_client, test_user_data, cleanup_user):
|
||||
"""测试分页获取用户并排序成功"""
|
||||
import time
|
||||
user_api = UserAPI(authenticated_client)
|
||||
|
||||
timestamp = int(time.time() * 1000)
|
||||
user1_data = test_user_data.copy()
|
||||
user1_data["username"] = f"user_a_{timestamp}"
|
||||
user1_data["email"] = f"user_a_{timestamp}@example.com"
|
||||
create_response1 = await user_api.create_user(user1_data)
|
||||
cleanup_user.append(create_response1.json()["id"])
|
||||
|
||||
user2_data = test_user_data.copy()
|
||||
user2_data["username"] = f"user_b_{timestamp}"
|
||||
user2_data["email"] = f"user_b_{timestamp}@example.com"
|
||||
create_response2 = await user_api.create_user(user2_data)
|
||||
cleanup_user.append(create_response2.json()["id"])
|
||||
|
||||
response = await user_api.get_users_by_page(page=0, size=10, sort="username", order="asc")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
usernames = [user["username"] for user in data["content"]]
|
||||
assert usernames == sorted(usernames)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_users_by_page_with_search(self, authenticated_client, test_user_data, cleanup_user):
|
||||
"""测试分页获取用户并搜索成功"""
|
||||
import time
|
||||
user_api = UserAPI(authenticated_client)
|
||||
|
||||
timestamp = int(time.time() * 1000)
|
||||
user1_data = test_user_data.copy()
|
||||
user1_data["username"] = f"search_test_user_{timestamp}"
|
||||
user1_data["email"] = f"search_test_{timestamp}@example.com"
|
||||
create_response1 = await user_api.create_user(user1_data)
|
||||
cleanup_user.append(create_response1.json()["id"])
|
||||
|
||||
user2_data = test_user_data.copy()
|
||||
user2_data["username"] = f"other_user_{timestamp}"
|
||||
user2_data["email"] = f"other_{timestamp}@example.com"
|
||||
create_response2 = await user_api.create_user(user2_data)
|
||||
cleanup_user.append(create_response2.json()["id"])
|
||||
|
||||
response = await user_api.get_users_by_page(page=0, size=10, keyword="search")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data["content"]) >= 1
|
||||
assert all("search" in user["username"] or "search" in user["email"]
|
||||
for user in data["content"])
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_user_count_success(self, authenticated_client, test_user_data, cleanup_user):
|
||||
"""测试获取用户总数成功"""
|
||||
user_api = UserAPI(authenticated_client)
|
||||
|
||||
initial_count_response = await user_api.get_user_count()
|
||||
initial_count = initial_count_response.json()
|
||||
|
||||
create_response = await user_api.create_user(test_user_data)
|
||||
cleanup_user.append(create_response.json()["id"])
|
||||
|
||||
final_count_response = await user_api.get_user_count()
|
||||
final_count = final_count_response.json()
|
||||
|
||||
assert final_count == initial_count + 1
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_users_by_page_with_different_page_sizes(self, authenticated_client, test_user_data, cleanup_user):
|
||||
"""测试不同页面大小的分页获取用户成功"""
|
||||
import time
|
||||
user_api = UserAPI(authenticated_client)
|
||||
|
||||
timestamp = int(time.time() * 1000)
|
||||
for i in range(15):
|
||||
user_data = test_user_data.copy()
|
||||
user_data["username"] = f"pagesize_test_{timestamp}_{i}"
|
||||
user_data["email"] = f"pagesize_test_{timestamp}_{i}@example.com"
|
||||
create_response = await user_api.create_user(user_data)
|
||||
cleanup_user.append(create_response.json()["id"])
|
||||
|
||||
response_size_10 = await user_api.get_users_by_page(page=0, size=10)
|
||||
assert response_size_10.status_code == 200
|
||||
data_size_10 = response_size_10.json()
|
||||
assert len(data_size_10["content"]) == 10
|
||||
|
||||
response_size_5 = await user_api.get_users_by_page(page=0, size=5)
|
||||
assert response_size_5.status_code == 200
|
||||
data_size_5 = response_size_5.json()
|
||||
assert len(data_size_5["content"]) == 5
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_users_by_page_with_page_navigation(self, authenticated_client, test_user_data, cleanup_user):
|
||||
"""测试分页导航成功"""
|
||||
import time
|
||||
user_api = UserAPI(authenticated_client)
|
||||
|
||||
timestamp = int(time.time() * 1000)
|
||||
for i in range(25):
|
||||
user_data = test_user_data.copy()
|
||||
user_data["username"] = f"pagination_test_{timestamp}_{i}"
|
||||
user_data["email"] = f"pagination_test_{timestamp}_{i}@example.com"
|
||||
create_response = await user_api.create_user(user_data)
|
||||
cleanup_user.append(create_response.json()["id"])
|
||||
|
||||
page1_response = await user_api.get_users_by_page(page=0, size=10)
|
||||
page1_data = page1_response.json()
|
||||
assert page1_data["currentPage"] == 0
|
||||
assert len(page1_data["content"]) == 10
|
||||
|
||||
page2_response = await user_api.get_users_by_page(page=1, size=10)
|
||||
page2_data = page2_response.json()
|
||||
assert page2_data["currentPage"] == 1
|
||||
assert len(page2_data["content"]) == 10
|
||||
|
||||
page3_response = await user_api.get_users_by_page(page=2, size=10)
|
||||
page3_data = page3_response.json()
|
||||
assert page3_data["currentPage"] == 2
|
||||
assert len(page3_data["content"]) >= 5
|
||||
|
||||
Reference in New Issue
Block a user