张翔
800 lines
30 KiB
Python
800 lines
30 KiB
Python
"""
|
|
comprehensive E2E测试套件
|
|
|
|
测试范围：
|
|
1. 用户管理完整生命周期
|
|
2. 角色管理完整生命周期
|
|
3. 菜单管理完整生命周期
|
|
4. 权限管理完整生命周期
|
|
5. 字典管理完整生命周期
|
|
6. 系统配置管理
|
|
7. 通知管理
|
|
8. 文件管理
|
|
9. 审计日志
|
|
10. 多角色多用户复杂场景
|
|
11. 并发操作测试
|
|
12. 错误恢复测试
|
|
"""
|
|
|
|
import pytest
|
|
import time
|
|
import uuid
|
|
import asyncio
|
|
from typing import Dict, Any
|
|
from api.auth_api import AuthAPI
|
|
from api.user_api import UserAPI
|
|
from api.role_api import RoleAPI
|
|
from api.menu_api import MenuAPI
|
|
from api.dict_api import DictAPI
|
|
from api.config_api import ConfigAPI
|
|
from api.notice_api import SysNoticeAPI
|
|
from api.file_api import FileAPI
|
|
from api.audit_api import AuditAPI
|
|
from config.settings import settings
|
|
|
|
|
|
@pytest.mark.e2e
|
|
@pytest.mark.comprehensive
|
|
@pytest.mark.regression
|
|
class TestComprehensiveE2E:
|
|
"""综合端到端测试类"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_role_menu_permission_full_lifecycle(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试用户-角色-菜单-权限完整生命周期"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
menu_api = MenuAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 创建测试角色
|
|
role_data = {
|
|
"roleName": f"Comprehensive_Role_{unique_id}",
|
|
"roleKey": f"comprehensive_role_{unique_id}",
|
|
"roleSort": 1,
|
|
"status": 1
|
|
}
|
|
role_response = await role_api.create_role(role_data)
|
|
assert role_response.status_code == 201
|
|
role_id = role_response.json()["id"]
|
|
test_data_manager.add_role(role_id)
|
|
|
|
# 2. 创建测试菜单
|
|
menu_data = {
|
|
"parentId": 0,
|
|
"menuName": f"Comprehensive_Menu_{unique_id}",
|
|
"menuType": "M",
|
|
"orderNum": 1,
|
|
"component": "Layout",
|
|
"perms": f"comprehensive:{unique_id}",
|
|
"status": 1
|
|
}
|
|
menu_response = await menu_api.create_menu(menu_data)
|
|
assert menu_response.status_code == 201
|
|
menu_id = menu_response.json()["id"]
|
|
test_data_manager.add_menu(menu_id)
|
|
|
|
# 3. 创建测试用户
|
|
user_data = {
|
|
"username": f"comprehensive_user_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"comprehensive_{unique_id}@example.com",
|
|
"roleId": role_id,
|
|
"status": 1
|
|
}
|
|
user_response = await user_api.create_user(user_data)
|
|
assert user_response.status_code == 201
|
|
user_id = user_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# 4. 分配菜单权限给角色
|
|
permission_data = {"menuIds": [menu_id]}
|
|
permission_response = await role_api.assign_permissions(role_id, permission_data)
|
|
assert permission_response.status_code == 200
|
|
|
|
# 5. 验证用户可以获取菜单
|
|
menus_response = await menu_api.get_user_menus(user_id)
|
|
assert menus_response.status_code == 200
|
|
menus = menus_response.json()
|
|
assert any(m["id"] == menu_id for m in menus)
|
|
|
|
# 6. 更新用户信息
|
|
update_data = {"email": f"updated_{unique_id}@example.com"}
|
|
update_response = await user_api.update_user(user_id, update_data)
|
|
assert update_response.status_code == 200
|
|
|
|
# 7. 更新角色信息
|
|
role_update_data = {"roleName": f"Updated_Role_{unique_id}"}
|
|
role_update_response = await role_api.update_role(role_id, role_update_data)
|
|
assert role_update_response.status_code == 200
|
|
|
|
# 8. 更新菜单信息
|
|
menu_update_data = {"menuName": f"Updated_Menu_{unique_id}"}
|
|
menu_update_response = await menu_api.update_menu(menu_id, menu_update_data)
|
|
assert menu_update_response.status_code == 200
|
|
|
|
# 9. 删除权限分配
|
|
await role_api.assign_permissions(role_id, {"menuIds": []})
|
|
|
|
# 10. 删除用户
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
# 11. 删除角色
|
|
await role_api.delete_role(role_id)
|
|
test_data_manager._roles.remove(role_id)
|
|
|
|
# 12. 删除菜单
|
|
await menu_api.delete_menu(menu_id)
|
|
test_data_manager._menus.remove(menu_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_dictionary_and_config_full_lifecycle(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试字典和系统配置完整生命周期"""
|
|
dict_api = DictAPI(authenticated_client)
|
|
config_api = ConfigAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 创建字典类型
|
|
dict_type_data = {
|
|
"type": f"TEST_TYPE_{unique_id}",
|
|
"name": f"测试类型_{unique_id}",
|
|
"remark": "E2E测试字典类型",
|
|
"sort": 1
|
|
}
|
|
dict_type_response = await dict_api.create_type(dict_type_data)
|
|
assert dict_type_response.status_code == 201
|
|
dict_type_id = dict_type_response.json()["id"]
|
|
test_data_manager.add_dict_type(dict_type_id)
|
|
|
|
# 2. 创建字典数据
|
|
dict_data = {
|
|
"type": f"TEST_TYPE_{unique_id}",
|
|
"code": f"TEST_CODE_{unique_id}",
|
|
"name": f"测试数据_{unique_id}",
|
|
"value": "1",
|
|
"remark": "E2E测试字典数据",
|
|
"sort": 1
|
|
}
|
|
dict_response = await dict_api.create(dict_data)
|
|
assert dict_response.status_code == 201
|
|
dict_id = dict_response.json()["id"]
|
|
test_data_manager.add_dict(dict_id)
|
|
|
|
# 3. 创建系统配置
|
|
config_data = {
|
|
"configKey": f"test_key_{unique_id}",
|
|
"configName": f"测试配置_{unique_id}",
|
|
"configType": "Y",
|
|
"configValue": "test_value",
|
|
"remark": "E2E测试配置"
|
|
}
|
|
config_response = await config_api.create_config(config_data)
|
|
assert config_response.status_code == 201
|
|
config_id = config_response.json()["id"]
|
|
test_data_manager.add_config(config_id)
|
|
|
|
# 4. 验证字典类型
|
|
type_get_response = await dict_api.get_type_by_id(dict_type_id)
|
|
assert type_get_response.status_code == 200
|
|
|
|
# 5. 验证字典数据
|
|
data_get_response = await dict_api.get_dict_by_id(dict_id)
|
|
assert data_get_response.status_code == 200
|
|
|
|
# 6. 验证系统配置
|
|
config_get_response = await config_api.get_config_by_id(config_id)
|
|
assert config_get_response.status_code == 200
|
|
|
|
# 7. 更新字典类型
|
|
type_update_data = {"name": f"更新类型_{unique_id}"}
|
|
type_update_response = await dict_api.update_type(dict_type_id, type_update_data)
|
|
assert type_update_response.status_code == 200
|
|
|
|
# 8. 更新字典数据
|
|
data_update_data = {"name": f"更新数据_{unique_id}"}
|
|
data_update_response = await dict_api.update_dict(dict_id, data_update_data)
|
|
assert data_update_response.status_code == 200
|
|
|
|
# 9. 更新系统配置
|
|
config_update_data = {"configName": f"更新配置_{unique_id}"}
|
|
config_update_response = await config_api.update_config(config_id, config_update_data)
|
|
assert config_update_response.status_code == 200
|
|
|
|
# 10. 删除字典数据
|
|
await dict_api.delete_dict(dict_id)
|
|
test_data_manager._dicts.remove(dict_id)
|
|
|
|
# 11. 删除字典类型
|
|
await dict_api.delete_type(dict_type_id)
|
|
test_data_manager._dict_types.remove(dict_type_id)
|
|
|
|
# 12. 删除系统配置
|
|
await config_api.delete_config(config_id)
|
|
test_data_manager._configs.remove(config_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_notice_and_file_full_lifecycle(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试通知和文件管理完整生命周期"""
|
|
notice_api = SysNoticeAPI(authenticated_client)
|
|
file_api = FileAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 创建通知
|
|
notice_data = {
|
|
"noticeTitle": f"E2E_Notice_{unique_id}",
|
|
"noticeType": "1",
|
|
"noticeContent": "This is an E2E test notice for comprehensive testing",
|
|
"status": "0"
|
|
}
|
|
notice_response = await notice_api.create(notice_data)
|
|
assert notice_response.status_code in [200, 201]
|
|
notice_data_response = notice_response.json()
|
|
|
|
notice_id = notice_data_response.get("id")
|
|
if not notice_id:
|
|
notice_title = notice_data_response.get("noticeTitle")
|
|
all_notices = await notice_api.get_all()
|
|
notices = all_notices.json()
|
|
notice = next((n for n in notices if n["noticeTitle"] == notice_title), None)
|
|
notice_id = notice["id"] if notice else None
|
|
|
|
assert notice_id is not None
|
|
test_data_manager.add_notice(notice_id)
|
|
|
|
# 2. 验证通知
|
|
notice_get_response = await notice_api.get_by_id(notice_id)
|
|
assert notice_get_response.status_code == 200
|
|
|
|
# 3. 更新通知
|
|
notice_update_data = {"noticeTitle": f"Updated_Notice_{unique_id}"}
|
|
notice_update_response = await notice_api.update(notice_id, notice_update_data)
|
|
assert notice_update_response.status_code == 200
|
|
|
|
# 4. 上传文件
|
|
file_response = await file_api.upload_file(
|
|
"test_file.txt",
|
|
b"This is a test file content for E2E testing"
|
|
)
|
|
assert file_response.status_code == 200
|
|
file_data = file_response.json()
|
|
file_id = file_data.get("id") or file_data.get("fileId")
|
|
|
|
if file_id:
|
|
test_data_manager.add_file(file_id)
|
|
|
|
# 5. 验证文件列表
|
|
file_list_response = await file_api.get_file_list(page=0, size=10)
|
|
assert file_list_response.status_code == 200
|
|
|
|
# 6. 删除通知
|
|
await notice_api.delete(notice_id)
|
|
test_data_manager._notices.remove(notice_id)
|
|
|
|
# 7. 删除文件（如果存在）
|
|
if file_id:
|
|
await file_api.delete_file(file_id)
|
|
if hasattr(test_data_manager, '_files'):
|
|
test_data_manager._files.remove(file_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_audit_log_full_lifecycle(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试审计日志完整生命周期"""
|
|
audit_api = AuditAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 创建测试用户以触发审计日志
|
|
user_data = {
|
|
"username": f"audit_user_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"audit_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
user_response = await UserAPI(authenticated_client).create_user(user_data)
|
|
assert user_response.status_code == 201
|
|
user_id = user_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# 2. 获取操作日志
|
|
operation_log_response = await audit_api.get_operation_logs(
|
|
page=0, size=10, operation=f"audit_user_{unique_id}"
|
|
)
|
|
assert operation_log_response.status_code == 200
|
|
|
|
# 3. 获取登录日志
|
|
login_log_response = await audit_api.get_login_logs(page=0, size=10)
|
|
assert login_log_response.status_code == 200
|
|
|
|
# 4. 获取异常日志
|
|
exception_log_response = await audit_api.get_exception_logs(page=0, size=10)
|
|
assert exception_log_response.status_code == 200
|
|
|
|
# 5. 清理测试用户
|
|
await UserAPI(authenticated_client).delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multi_user_role_concurrent_operations(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试多用户多角色并发操作"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 创建多个角色
|
|
roles = []
|
|
for i in range(3):
|
|
role_data = {
|
|
"roleName": f"Concurrent_Role_{unique_id}_{i}",
|
|
"roleKey": f"concurrent_role_{unique_id}_{i}",
|
|
"roleSort": i + 1,
|
|
"status": 1
|
|
}
|
|
role_response = await role_api.create_role(role_data)
|
|
assert role_response.status_code == 201
|
|
role_id = role_response.json()["id"]
|
|
roles.append(role_id)
|
|
test_data_manager.add_role(role_id)
|
|
|
|
# 创建多个用户
|
|
users = []
|
|
for i in range(5):
|
|
user_data = {
|
|
"username": f"concurrent_user_{unique_id}_{i}",
|
|
"password": "Test123!@#",
|
|
"email": f"concurrent_{unique_id}_{i}@example.com",
|
|
"roleId": roles[i % 3],
|
|
"status": 1
|
|
}
|
|
user_response = await user_api.create_user(user_data)
|
|
assert user_response.status_code == 201
|
|
user_id = user_response.json()["id"]
|
|
users.append(user_id)
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# 并发更新用户
|
|
for i, user_id in enumerate(users):
|
|
update_data = {"email": f"updated_{unique_id}_{i}@example.com"}
|
|
update_response = await user_api.update_user(user_id, update_data)
|
|
assert update_response.status_code == 200
|
|
|
|
# 并发更新角色
|
|
for i, role_id in enumerate(roles):
|
|
role_update_data = {"roleSort": len(roles) - i}
|
|
role_update_response = await role_api.update_role(role_id, role_update_data)
|
|
assert role_update_response.status_code == 200
|
|
|
|
# 验证所有用户和角色
|
|
for user_id in users:
|
|
user_response = await user_api.get_user_by_id(user_id)
|
|
assert user_response.status_code == 200
|
|
|
|
for role_id in roles:
|
|
role_response = await role_api.get_role_by_id(role_id)
|
|
assert role_response.status_code == 200
|
|
|
|
# 清理
|
|
for user_id in users:
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
for role_id in roles:
|
|
await role_api.delete_role(role_id)
|
|
test_data_manager._roles.remove(role_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_error_recovery_and_validation(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试错误恢复和验证"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 测试无效输入
|
|
invalid_user_data = {
|
|
"username": "",
|
|
"password": "123",
|
|
"email": "invalid-email"
|
|
}
|
|
invalid_response = await user_api.create_user(invalid_user_data)
|
|
assert invalid_response.status_code in [400, 422]
|
|
|
|
invalid_role_data = {
|
|
"roleName": "",
|
|
"roleKey": "",
|
|
"roleSort": 0
|
|
}
|
|
invalid_role_response = await role_api.create_role(invalid_role_data)
|
|
assert invalid_role_response.status_code in [400, 422]
|
|
|
|
# 2. 测试重复数据
|
|
user_data = {
|
|
"username": f"recovery_user_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"recovery_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
first_response = await user_api.create_user(user_data)
|
|
assert first_response.status_code == 201
|
|
user_id = first_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
second_response = await user_api.create_user(user_data)
|
|
assert second_response.status_code in [400, 409]
|
|
|
|
# 3. 测试获取不存在的数据
|
|
not_found_response = await user_api.get_user_by_id(999999)
|
|
assert not_found_response.status_code in [404, 500]
|
|
|
|
# 4. 测试更新不存在的数据
|
|
update_not_found_response = await user_api.update_user(
|
|
999999, {"email": "test@example.com"}
|
|
)
|
|
assert update_not_found_response.status_code in [404, 500]
|
|
|
|
# 5. 测试删除不存在的数据
|
|
delete_not_found_response = await user_api.delete_user(999999)
|
|
assert delete_not_found_response.status_code in [204, 404, 500]
|
|
|
|
# 6. 清理
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pagination_and_filtering(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试分页和过滤"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 创建多个用户
|
|
user_ids = []
|
|
for i in range(15):
|
|
user_data = {
|
|
"username": f"pagination_user_{unique_id}_{i}",
|
|
"password": "Test123!@#",
|
|
"email": f"pagination_{unique_id}_{i}@example.com",
|
|
"status": 1
|
|
}
|
|
user_response = await user_api.create_user(user_data)
|
|
assert user_response.status_code == 201
|
|
user_ids.append(user_response.json()["id"])
|
|
test_data_manager.add_user(user_ids[-1])
|
|
|
|
# 测试不同页面大小
|
|
for page_size in [5, 10, 20]:
|
|
response = await user_api.get_users_by_page(page=0, size=page_size)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "content" in data
|
|
assert "totalElements" in data
|
|
assert len(data["content"]) <= page_size
|
|
|
|
# 测试分页导航
|
|
page1 = await user_api.get_users_by_page(page=0, size=5)
|
|
page2 = await user_api.get_users_by_page(page=1, size=5)
|
|
|
|
assert page1.status_code == 200
|
|
assert page2.status_code == 200
|
|
|
|
page1_data = page1.json()
|
|
page2_data = page2.json()
|
|
|
|
assert page1_data["currentPage"] == 0
|
|
assert page2_data["currentPage"] == 1
|
|
assert page1_data["totalPages"] >= 2
|
|
|
|
# 测试搜索
|
|
search_response = await user_api.get_users_by_page(
|
|
page=0, size=10, keyword=f"pagination_user_{unique_id}"
|
|
)
|
|
assert search_response.status_code == 200
|
|
search_data = search_response.json()
|
|
assert len(search_data["content"]) >= 1
|
|
|
|
# 测试排序
|
|
sort_response = await user_api.get_users_by_page(
|
|
page=0, size=10, sort="username", order="asc"
|
|
)
|
|
assert sort_response.status_code == 200
|
|
|
|
# 清理
|
|
for user_id in user_ids:
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_data_integrity_and_consistency(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试数据完整性和一致性"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 创建角色
|
|
role_data = {
|
|
"roleName": f"Integrity_Role_{unique_id}",
|
|
"roleKey": f"integrity_role_{unique_id}",
|
|
"roleSort": 1,
|
|
"status": 1
|
|
}
|
|
role_response = await role_api.create_role(role_data)
|
|
assert role_response.status_code == 201
|
|
role_id = role_response.json()["id"]
|
|
test_data_manager.add_role(role_id)
|
|
|
|
# 2. 创建用户并关联角色
|
|
user_data = {
|
|
"username": f"integrity_user_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"integrity_{unique_id}@example.com",
|
|
"roleId": role_id,
|
|
"status": 1
|
|
}
|
|
user_response = await user_api.create_user(user_data)
|
|
assert user_response.status_code == 201
|
|
user_id = user_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# 3. 验证用户角色关联
|
|
user_get_response = await user_api.get_user_by_id(user_id)
|
|
assert user_get_response.status_code == 200
|
|
user_data_result = user_get_response.json()
|
|
assert user_data_result["roleId"] == role_id
|
|
|
|
# 4. 更新角色并验证用户数据不变
|
|
role_update_data = {"roleName": f"Updated_Integrity_Role_{unique_id}"}
|
|
await role_api.update_role(role_id, role_update_data)
|
|
|
|
user_verify_response = await user_api.get_user_by_id(user_id)
|
|
assert user_verify_response.json()["roleId"] == role_id
|
|
|
|
# 5. 删除用户
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
# 6. 删除角色
|
|
await role_api.delete_role(role_id)
|
|
test_data_manager._roles.remove(role_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_performance_and_stress(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试性能和压力"""
|
|
user_api = UserAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 1. 批量创建用户
|
|
start_time = time.time()
|
|
user_ids = []
|
|
|
|
for i in range(50):
|
|
user_data = {
|
|
"username": f"stress_user_{unique_id}_{i}",
|
|
"password": "Test123!@#",
|
|
"email": f"stress_{unique_id}_{i}@example.com",
|
|
"status": 1
|
|
}
|
|
user_response = await user_api.create_user(user_data)
|
|
if user_response.status_code == 201:
|
|
user_ids.append(user_response.json()["id"])
|
|
test_data_manager.add_user(user_ids[-1])
|
|
|
|
create_duration = time.time() - start_time
|
|
print(f"批量创建50个用户耗时: {create_duration:.2f}秒")
|
|
|
|
# 2. 批量获取用户
|
|
start_time = time.time()
|
|
for user_id in user_ids[:20]:
|
|
response = await user_api.get_user_by_id(user_id)
|
|
assert response.status_code == 200
|
|
|
|
get_duration = time.time() - start_time
|
|
print(f"批量获取20个用户耗时: {get_duration:.2f}秒")
|
|
|
|
# 3. 验证性能指标
|
|
assert create_duration < 30, f"创建50个用户耗时过长: {create_duration:.2f}秒"
|
|
assert get_duration < 10, f"获取20个用户耗时过长: {get_duration:.2f}秒"
|
|
|
|
# 4. 清理
|
|
for user_id in user_ids:
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_complete_business_workflow(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""测试完整业务流程"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
menu_api = MenuAPI(authenticated_client)
|
|
dict_api = DictAPI(authenticated_client)
|
|
config_api = ConfigAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# ========== 1. 角色管理流程 ==========
|
|
role_data = {
|
|
"roleName": f"Workflow_Role_{unique_id}",
|
|
"roleKey": f"workflow_role_{unique_id}",
|
|
"roleSort": 1,
|
|
"status": 1
|
|
}
|
|
role_response = await role_api.create_role(role_data)
|
|
assert role_response.status_code == 201
|
|
role_id = role_response.json()["id"]
|
|
test_data_manager.add_role(role_id)
|
|
|
|
# ========== 2. 菜单管理流程 ==========
|
|
menu_data = {
|
|
"parentId": 0,
|
|
"menuName": f"Workflow_Menu_{unique_id}",
|
|
"menuType": "M",
|
|
"orderNum": 1,
|
|
"component": "Layout",
|
|
"perms": f"workflow:{unique_id}",
|
|
"status": 1
|
|
}
|
|
menu_response = await menu_api.create_menu(menu_data)
|
|
assert menu_response.status_code == 201
|
|
menu_id = menu_response.json()["id"]
|
|
test_data_manager.add_menu(menu_id)
|
|
|
|
# 分配权限
|
|
await role_api.assign_permissions(role_id, {"menuIds": [menu_id]})
|
|
|
|
# ========== 3. 用户管理流程 ==========
|
|
user_data = {
|
|
"username": f"workflow_user_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"workflow_{unique_id}@example.com",
|
|
"roleId": role_id,
|
|
"status": 1
|
|
}
|
|
user_response = await user_api.create_user(user_data)
|
|
assert user_response.status_code == 201
|
|
user_id = user_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# ========== 4. 字典管理流程 ==========
|
|
dict_type_data = {
|
|
"type": f"WORKFLOW_TYPE_{unique_id}",
|
|
"name": f"工作流类型_{unique_id}",
|
|
"remark": "业务流程测试",
|
|
"sort": 1
|
|
}
|
|
dict_type_response = await dict_api.create_type(dict_type_data)
|
|
assert dict_type_response.status_code == 201
|
|
dict_type_id = dict_type_response.json()["id"]
|
|
test_data_manager.add_dict_type(dict_type_id)
|
|
|
|
dict_data = {
|
|
"type": f"WORKFLOW_TYPE_{unique_id}",
|
|
"code": f"WORKFLOW_CODE_{unique_id}",
|
|
"name": f"工作流数据_{unique_id}",
|
|
"value": "1",
|
|
"remark": "业务流程测试数据",
|
|
"sort": 1
|
|
}
|
|
dict_response = await dict_api.create(dict_data)
|
|
assert dict_response.status_code == 201
|
|
dict_id = dict_response.json()["id"]
|
|
test_data_manager.add_dict(dict_id)
|
|
|
|
# ========== 5. 系统配置流程 ==========
|
|
config_data = {
|
|
"configKey": f"workflow_key_{unique_id}",
|
|
"configName": f"工作流配置_{unique_id}",
|
|
"configType": "Y",
|
|
"configValue": "workflow_value",
|
|
"remark": "业务流程测试配置"
|
|
}
|
|
config_response = await config_api.create_config(config_data)
|
|
assert config_response.status_code == 201
|
|
config_id = config_response.json()["id"]
|
|
test_data_manager.add_config(config_id)
|
|
|
|
# ========== 6. 更新流程 ==========
|
|
# 更新用户
|
|
update_user_data = {"email": f"updated_workflow_{unique_id}@example.com"}
|
|
await user_api.update_user(user_id, update_user_data)
|
|
|
|
# 更新角色
|
|
update_role_data = {"roleName": f"Updated_Workflow_Role_{unique_id}"}
|
|
await role_api.update_role(role_id, update_role_data)
|
|
|
|
# 更新菜单
|
|
update_menu_data = {"menuName": f"Updated_Workflow_Menu_{unique_id}"}
|
|
await menu_api.update_menu(menu_id, update_menu_data)
|
|
|
|
# 更新字典
|
|
update_dict_data = {"name": f"更新工作流数据_{unique_id}"}
|
|
await dict_api.update_dict(dict_id, update_dict_data)
|
|
|
|
# 更新配置
|
|
update_config_data = {"configName": f"更新工作流配置_{unique_id}"}
|
|
await config_api.update_config(config_id, update_config_data)
|
|
|
|
# ========== 7. 查询验证流程 ==========
|
|
# 验证用户
|
|
user_verify = await user_api.get_user_by_id(user_id)
|
|
assert user_verify.status_code == 200
|
|
|
|
# 验证角色
|
|
role_verify = await role_api.get_role_by_id(role_id)
|
|
assert role_verify.status_code == 200
|
|
|
|
# 验证菜单
|
|
menu_verify = await menu_api.get_menu_by_id(menu_id)
|
|
assert menu_verify.status_code == 200
|
|
|
|
# 验证字典
|
|
dict_verify = await dict_api.get_dict_by_id(dict_id)
|
|
assert dict_verify.status_code == 200
|
|
|
|
# 验证配置
|
|
config_verify = await config_api.get_config_by_id(config_id)
|
|
assert config_verify.status_code == 200
|
|
|
|
# ========== 8. 删除流程 ==========
|
|
# 删除配置
|
|
await config_api.delete_config(config_id)
|
|
test_data_manager._configs.remove(config_id)
|
|
|
|
# 删除字典
|
|
await dict_api.delete_dict(dict_id)
|
|
test_data_manager._dicts.remove(dict_id)
|
|
|
|
# 删除字典类型
|
|
await dict_api.delete_type(dict_type_id)
|
|
test_data_manager._dict_types.remove(dict_type_id)
|
|
|
|
# 删除用户
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
|
|
# 删除角色
|
|
await role_api.delete_role(role_id)
|
|
test_data_manager._roles.remove(role_id)
|
|
|
|
# 删除菜单
|
|
await menu_api.delete_menu(menu_id)
|
|
test_data_manager._menus.remove(menu_id)
|
|
|
|
# ========== 9. 删除后验证 ==========
|
|
# 验证用户已删除
|
|
user_deleted = await user_api.get_user_by_id(user_id)
|
|
assert user_deleted.status_code in [404, 200]
|
|
|
|
# 验证角色已删除
|
|
role_deleted = await role_api.get_role_by_id(role_id)
|
|
assert role_deleted.status_code in [404, 200]
|
|
|
|
# 验证菜单已删除
|
|
menu_deleted = await menu_api.get_menu_by_id(menu_id)
|
|
assert menu_deleted.status_code in [404, 200]
|