张翔
1279 lines
56 KiB
Python
1279 lines
56 KiB
Python
"""
|
|
UAT用户验收测试套件
|
|
|
|
测试范围：
|
|
1. 业务场景完整性验证
|
|
2. 用户角色权限配置验证
|
|
3. 系统配置验证
|
|
4. 数据字典验证
|
|
5. 通知公告验证
|
|
6. 文件管理验证
|
|
7. 审计日志验证
|
|
8. 用户体验验证
|
|
9. 性能验收验证
|
|
10. 安全验收验证
|
|
"""
|
|
|
|
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.uat
|
|
@pytest.mark.acceptance
|
|
@pytest.mark.regression
|
|
class TestUATAcceptance:
|
|
"""用户验收测试类 - 模拟真实业务场景"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_user_management_business_scenario(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-01: 用户管理业务场景验证
|
|
|
|
场景描述：
|
|
- 系统管理员创建新员工账户
|
|
- 分配相应角色和权限
|
|
- 员工登录系统并访问授权资源
|
|
- 员工信息变更处理
|
|
- 员工离职账户禁用
|
|
"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 创建普通员工角色
|
|
employee_role_data = {
|
|
"roleName": f"UAT_Employee_Role_{unique_id}",
|
|
"roleKey": f"uat_employee_role_{unique_id}",
|
|
"roleSort": 10,
|
|
"status": 1,
|
|
"remark": "UAT测试-普通员工角色"
|
|
}
|
|
employee_role_response = await role_api.create_role(employee_role_data)
|
|
assert employee_role_response.status_code == 201, "创建员工角色失败"
|
|
employee_role_id = employee_role_response.json()["id"]
|
|
test_data_manager.add_role(employee_role_id)
|
|
|
|
# 步骤2: 创建部门经理角色
|
|
manager_role_data = {
|
|
"roleName": f"UAT_Manager_Role_{unique_id}",
|
|
"roleKey": f"uat_manager_role_{unique_id}",
|
|
"roleSort": 9,
|
|
"status": 1,
|
|
"remark": "UAT测试-部门经理角色"
|
|
}
|
|
manager_role_response = await role_api.create_role(manager_role_data)
|
|
assert manager_role_response.status_code == 201, "创建经理角色失败"
|
|
manager_role_id = manager_role_response.json()["id"]
|
|
test_data_manager.add_role(manager_role_id)
|
|
|
|
# 步骤3: 创建新员工账户（模拟入职）
|
|
new_employee_data = {
|
|
"username": f"uat_employee_{unique_id}",
|
|
"password": "Employee123!@#",
|
|
"email": f"employee_{unique_id}@company.com",
|
|
"roleId": employee_role_id,
|
|
"status": 1,
|
|
"remark": "UAT测试-新入职员工"
|
|
}
|
|
new_employee_response = await user_api.create_user(new_employee_data)
|
|
assert new_employee_response.status_code == 201, "创建新员工账户失败"
|
|
employee_id = new_employee_response.json()["id"]
|
|
test_data_manager.add_user(employee_id)
|
|
|
|
# 步骤4: 验证员工可以登录
|
|
login_response = await AuthAPI(authenticated_client).login(
|
|
f"uat_employee_{unique_id}", "Employee123!@#"
|
|
)
|
|
assert login_response.status_code == 200, "员工登录失败"
|
|
login_data = login_response.json()
|
|
assert "token" in login_data, "登录响应缺少token"
|
|
|
|
# 步骤5: 员工获取个人资料（使用已知用户ID）
|
|
profile_response = await user_api.get_user_by_id(employee_id)
|
|
assert profile_response.status_code == 200, "获取个人资料失败"
|
|
profile_data = profile_response.json()
|
|
assert profile_data["username"] == f"uat_employee_{unique_id}", "个人资料不匹配"
|
|
|
|
# 步骤6: 员工修改个人信息（使用已知用户ID）
|
|
update_profile_data = {
|
|
"nickname": f"张三_{unique_id}",
|
|
"phone": f"1380013800{unique_id[-4:]}",
|
|
"remark": "UAT测试-更新个人信息"
|
|
}
|
|
update_profile_response = await user_api.update_user(employee_id, update_profile_data)
|
|
assert update_profile_response.status_code == 200, "更新个人资料失败"
|
|
|
|
# 步骤7: 员工晋升为部门经理（角色变更）
|
|
assign_roles_response = await user_api.assign_roles(employee_id, [manager_role_id])
|
|
assert assign_roles_response.status_code == 200, "角色变更失败"
|
|
|
|
# 步骤8: 验证角色变更生效
|
|
user_detail_response = await user_api.get_user_by_id(employee_id)
|
|
assert user_detail_response.status_code == 200, "获取用户详情失败"
|
|
user_detail = user_detail_response.json()
|
|
assert manager_role_id in user_detail["roles"], "角色变更未生效"
|
|
|
|
# 步骤9: 员工离职（禁用账户）
|
|
disable_data = {"status": 0}
|
|
disable_response = await user_api.update_user(employee_id, disable_data)
|
|
assert disable_response.status_code == 200, "禁用账户失败"
|
|
|
|
# 步骤10: 验证已禁用账户无法登录
|
|
disabled_login_response = await AuthAPI(authenticated_client).login(
|
|
f"uat_employee_{unique_id}", "Employee123!@#"
|
|
)
|
|
assert disabled_login_response.status_code != 200, "已禁用账户不应能登录"
|
|
|
|
# 清理
|
|
await user_api.delete_user(employee_id)
|
|
test_data_manager._users.remove(employee_id)
|
|
await role_api.delete_role(manager_role_id)
|
|
test_data_manager._roles.remove(manager_role_id)
|
|
await role_api.delete_role(employee_role_id)
|
|
test_data_manager._roles.remove(employee_role_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_role_permission_configuration(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-02: 角色权限配置验证
|
|
|
|
场景描述：
|
|
- 创建不同级别的角色（管理员、经理、普通员工）
|
|
- 为每个角色配置不同的菜单和数据权限
|
|
- 验证权限隔离
|
|
- 验证权限继承
|
|
"""
|
|
role_api = RoleAPI(authenticated_client)
|
|
menu_api = MenuAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_role_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 创建系统管理员角色
|
|
admin_role_data = {
|
|
"roleName": f"UAT_Admin_Role_{unique_id}",
|
|
"roleKey": f"uat_admin_role_{unique_id}",
|
|
"roleSort": 1,
|
|
"status": 1,
|
|
"remark": "UAT测试-系统管理员角色"
|
|
}
|
|
admin_role_response = await role_api.create_role(admin_role_data)
|
|
assert admin_role_response.status_code == 201, "创建管理员角色失败"
|
|
admin_role_id = admin_role_response.json()["id"]
|
|
test_data_manager.add_role(admin_role_id)
|
|
|
|
# 步骤2: 创建业务经理角色
|
|
manager_role_data = {
|
|
"roleName": f"UAT_Business_Role_{unique_id}",
|
|
"roleKey": f"uat_business_role_{unique_id}",
|
|
"roleSort": 5,
|
|
"status": 1,
|
|
"remark": "UAT测试-业务经理角色"
|
|
}
|
|
manager_role_response = await role_api.create_role(manager_role_data)
|
|
assert manager_role_response.status_code == 201, "创建业务经理角色失败"
|
|
manager_role_id = manager_role_response.json()["id"]
|
|
test_data_manager.add_role(manager_role_id)
|
|
|
|
# 步骤3: 创建普通员工角色
|
|
employee_role_data = {
|
|
"roleName": f"UAT_Staff_Role_{unique_id}",
|
|
"roleKey": f"uat_staff_role_{unique_id}",
|
|
"roleSort": 10,
|
|
"status": 1,
|
|
"remark": "UAT测试-普通员工角色"
|
|
}
|
|
employee_role_response = await role_api.create_role(employee_role_data)
|
|
assert employee_role_response.status_code == 201, "创建普通员工角色失败"
|
|
employee_role_id = employee_role_response.json()["id"]
|
|
test_data_manager.add_role(employee_role_id)
|
|
|
|
# 步骤4: 创建系统管理菜单
|
|
system_menu_data = {
|
|
"parentId": 0,
|
|
"menuName": f"系统管理_{unique_id}",
|
|
"menuType": "M",
|
|
"orderNum": 1,
|
|
"component": "Layout",
|
|
"perms": f"system:{unique_id}",
|
|
"status": 1,
|
|
"icon": "system"
|
|
}
|
|
system_menu_response = await menu_api.create_menu(system_menu_data)
|
|
assert system_menu_response.status_code == 201, "创建系统管理菜单失败"
|
|
system_menu_id = system_menu_response.json()["id"]
|
|
test_data_manager.add_menu(system_menu_id)
|
|
|
|
# 步骤5: 创建用户管理子菜单
|
|
user_menu_data = {
|
|
"parentId": system_menu_id,
|
|
"menuName": "用户管理",
|
|
"menuType": "C",
|
|
"orderNum": 1,
|
|
"component": "UserManagement",
|
|
"perms": f"user:manage:{unique_id}",
|
|
"status": 1
|
|
}
|
|
user_menu_response = await menu_api.create_menu(user_menu_data)
|
|
assert user_menu_response.status_code == 201, "创建用户管理菜单失败"
|
|
user_menu_id = user_menu_response.json()["id"]
|
|
test_data_manager.add_menu(user_menu_id)
|
|
|
|
# 步骤6: 创建角色管理子菜单
|
|
role_menu_data = {
|
|
"parentId": system_menu_id,
|
|
"menuName": "角色管理",
|
|
"menuType": "C",
|
|
"orderNum": 2,
|
|
"component": "RoleManagement",
|
|
"perms": f"role:manage:{unique_id}",
|
|
"status": 1
|
|
}
|
|
role_menu_response = await menu_api.create_menu(role_menu_data)
|
|
assert role_menu_response.status_code == 201, "创建角色管理菜单失败"
|
|
role_menu_id = role_menu_response.json()["id"]
|
|
test_data_manager.add_menu(role_menu_id)
|
|
|
|
# 步骤7: 为管理员角色分配所有菜单权限
|
|
admin_permission_data = {"menuIds": [system_menu_id, user_menu_id, role_menu_id]}
|
|
admin_permission_response = await role_api.assign_permissions(admin_role_id, admin_permission_data)
|
|
assert admin_permission_response.status_code == 200, "为管理员分配权限失败"
|
|
|
|
# 步骤8: 为业务经理角色分配部分菜单权限
|
|
manager_permission_data = {"menuIds": [user_menu_id]}
|
|
manager_permission_response = await role_api.assign_permissions(manager_role_id, manager_permission_data)
|
|
assert manager_permission_response.status_code == 200, "为业务经理分配权限失败"
|
|
|
|
# 步骤9: 为普通员工角色分配最小权限
|
|
employee_permission_data = {"menuIds": []}
|
|
employee_permission_response = await role_api.assign_permissions(employee_role_id, employee_permission_data)
|
|
assert employee_permission_response.status_code == 200, "为普通员工分配权限失败"
|
|
|
|
# 步骤10: 验证管理员可以访问所有菜单
|
|
admin_menus_response = await menu_api.get_user_menus_by_role(admin_role_id)
|
|
assert admin_menus_response.status_code == 200, "获取管理员菜单失败"
|
|
admin_menus = admin_menus_response.json()
|
|
assert len(admin_menus) >= 3, "管理员应能访问至少3个菜单"
|
|
|
|
# 步骤11: 验证业务经理只能访问授权菜单
|
|
manager_menus_response = await menu_api.get_user_menus_by_role(manager_role_id)
|
|
assert manager_menus_response.status_code == 200, "获取业务经理菜单失败"
|
|
manager_menus = manager_menus_response.json()
|
|
assert len(manager_menus) == 1, "业务经理应只能访问1个菜单"
|
|
assert manager_menus[0]["id"] == user_menu_id, "业务经理菜单权限不正确"
|
|
|
|
# 步骤12: 验证普通员工无菜单权限
|
|
employee_menus_response = await menu_api.get_user_menus_by_role(employee_role_id)
|
|
assert employee_menus_response.status_code == 200, "获取普通员工菜单失败"
|
|
employee_menus = employee_menus_response.json()
|
|
assert len(employee_menus) == 0, "普通员工应无菜单权限"
|
|
|
|
# 清理
|
|
await role_api.assign_permissions(admin_role_id, {"menuIds": []})
|
|
await role_api.assign_permissions(manager_role_id, {"menuIds": []})
|
|
await role_api.assign_permissions(employee_role_id, {"menuIds": []})
|
|
|
|
await menu_api.delete_menu(role_menu_id)
|
|
test_data_manager._menus.remove(role_menu_id)
|
|
await menu_api.delete_menu(user_menu_id)
|
|
test_data_manager._menus.remove(user_menu_id)
|
|
await menu_api.delete_menu(system_menu_id)
|
|
test_data_manager._menus.remove(system_menu_id)
|
|
|
|
await role_api.delete_role(employee_role_id)
|
|
test_data_manager._roles.remove(employee_role_id)
|
|
await role_api.delete_role(manager_role_id)
|
|
test_data_manager._roles.remove(manager_role_id)
|
|
await role_api.delete_role(admin_role_id)
|
|
test_data_manager._roles.remove(admin_role_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_system_configuration_validation(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-03: 系统配置验证
|
|
|
|
场景描述：
|
|
- 修改系统基本配置
|
|
- 验证配置生效
|
|
- 修改邮件配置
|
|
- 验证配置有效性
|
|
- 配置回滚
|
|
"""
|
|
config_api = ConfigAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_config_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 获取当前系统配置
|
|
config_list_response = await config_api.get_config_list(page=0, size=100)
|
|
assert config_list_response.status_code == 200, "获取系统配置列表失败"
|
|
config_list = config_list_response.json()
|
|
|
|
# 步骤2: 创建测试配置项
|
|
test_config_data = {
|
|
"configKey": f"uat_test_key_{unique_id}",
|
|
"configName": f"UAT测试配置_{unique_id}",
|
|
"configType": "Y",
|
|
"configValue": "test_value_initial",
|
|
"remark": "UAT测试-系统配置验证"
|
|
}
|
|
test_config_response = await config_api.create_config(test_config_data)
|
|
assert test_config_response.status_code == 201, "创建测试配置失败"
|
|
test_config_id = test_config_response.json()["id"]
|
|
test_data_manager.add_config(test_config_id)
|
|
|
|
# 步骤3: 验证配置创建成功
|
|
get_config_response = await config_api.get_config_by_key(f"uat_test_key_{unique_id}")
|
|
assert get_config_response.status_code == 200, "获取测试配置失败"
|
|
get_config_data = get_config_response.json()
|
|
assert get_config_data["configValue"] == "test_value_initial", "配置值不匹配"
|
|
|
|
# 步骤4: 更新配置值
|
|
update_config_data = {
|
|
"configValue": "test_value_updated",
|
|
"remark": "UAT测试-更新配置值"
|
|
}
|
|
update_config_response = await config_api.update_config(test_config_id, update_config_data)
|
|
assert update_config_response.status_code == 200, "更新配置失败"
|
|
|
|
# 步骤5: 验证配置更新生效
|
|
verify_config_response = await config_api.get_config_by_key(f"uat_test_key_{unique_id}")
|
|
assert verify_config_response.status_code == 200, "验证配置更新失败"
|
|
verify_config_data = verify_config_response.json()
|
|
assert verify_config_data["configValue"] == "test_value_updated", "配置更新未生效"
|
|
|
|
# 步骤6: 创建邮件配置
|
|
email_config_data = {
|
|
"configKey": f"uat_email_config_{unique_id}",
|
|
"configName": "UAT测试邮件配置",
|
|
"configType": "Y",
|
|
"configValue": '{"host":"smtp.test.com","port":587,"username":"test@test.com","password":"test123"}',
|
|
"remark": "UAT测试-邮件配置"
|
|
}
|
|
email_config_response = await config_api.create_config(email_config_data)
|
|
assert email_config_response.status_code == 201, "创建邮件配置失败"
|
|
email_config_id = email_config_response.json()["id"]
|
|
test_data_manager.add_config(email_config_id)
|
|
|
|
# 步骤7: 验证邮件配置格式
|
|
email_config_get_response = await config_api.get_config_by_key(f"uat_email_config_{unique_id}")
|
|
assert email_config_get_response.status_code == 200, "获取邮件配置失败"
|
|
email_config = email_config_get_response.json()
|
|
import json
|
|
email_config_value = json.loads(email_config["configValue"])
|
|
assert "host" in email_config_value, "邮件配置缺少host字段"
|
|
assert "port" in email_config_value, "邮件配置缺少port字段"
|
|
|
|
# 步骤8: 修改邮件配置
|
|
updated_email_config = {
|
|
"configValue": '{"host":"smtp.updated.com","port":465,"username":"updated@test.com","password":"updated123"}',
|
|
"remark": "UAT测试-更新邮件配置"
|
|
}
|
|
updated_email_config_response = await config_api.update_config(email_config_id, updated_email_config)
|
|
assert updated_email_config_response.status_code == 200, "更新邮件配置失败"
|
|
|
|
# 步骤9: 验证邮件配置更新
|
|
verify_email_config_response = await config_api.get_config_by_key(f"uat_email_config_{unique_id}")
|
|
assert verify_email_config_response.status_code == 200, "验证邮件配置更新失败"
|
|
verify_email_config = verify_email_config_response.json()
|
|
verify_email_value = json.loads(verify_email_config["configValue"])
|
|
assert verify_email_value["host"] == "smtp.updated.com", "邮件配置更新未生效"
|
|
|
|
# 步骤10: 删除测试配置
|
|
await config_api.delete_config(test_config_id)
|
|
test_data_manager._configs.remove(test_config_id)
|
|
|
|
# 步骤11: 删除邮件配置
|
|
await config_api.delete_config(email_config_id)
|
|
test_data_manager._configs.remove(email_config_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_dictionary_management(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-04: 数据字典验证
|
|
|
|
场景描述：
|
|
- 创建字典类型
|
|
- 添加字典数据
|
|
- 修改字典数据
|
|
- 删除字典数据
|
|
- 验证字典缓存
|
|
"""
|
|
dict_api = DictAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_dict_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 创建用户状态字典类型
|
|
status_type_data = {
|
|
"type": f"USER_STATUS_{unique_id}",
|
|
"name": "用户状态",
|
|
"remark": "UAT测试-用户状态字典",
|
|
"sort": 1
|
|
}
|
|
status_type_response = await dict_api.create_type(status_type_data)
|
|
assert status_type_response.status_code == 201, "创建用户状态字典类型失败"
|
|
status_type_id = status_type_response.json()["id"]
|
|
test_data_manager.add_dict_type(status_type_id)
|
|
|
|
# 步骤2: 添加启用状态数据
|
|
enabled_status_data = {
|
|
"type": f"USER_STATUS_{unique_id}",
|
|
"code": "ENABLED",
|
|
"name": "启用",
|
|
"value": "1",
|
|
"remark": "用户已启用",
|
|
"sort": 1
|
|
}
|
|
enabled_status_response = await dict_api.create(enabled_status_data)
|
|
assert enabled_status_response.status_code == 201, "添加启用状态失败"
|
|
enabled_status_id = enabled_status_response.json()["id"]
|
|
test_data_manager.add_dict(enabled_status_id)
|
|
|
|
# 步骤3: 添加禁用状态数据
|
|
disabled_status_data = {
|
|
"type": f"USER_STATUS_{unique_id}",
|
|
"code": "DISABLED",
|
|
"name": "禁用",
|
|
"value": "0",
|
|
"remark": "用户已禁用",
|
|
"sort": 2
|
|
}
|
|
disabled_status_response = await dict_api.create(disabled_status_data)
|
|
assert disabled_status_response.status_code == 201, "添加禁用状态失败"
|
|
disabled_status_id = disabled_status_response.json()["id"]
|
|
test_data_manager.add_dict(disabled_status_id)
|
|
|
|
# 步骤4: 创建用户类型字典
|
|
user_type_data = {
|
|
"type": f"USER_TYPE_{unique_id}",
|
|
"name": "用户类型",
|
|
"remark": "UAT测试-用户类型字典",
|
|
"sort": 2
|
|
}
|
|
user_type_response = await dict_api.create_type(user_type_data)
|
|
assert user_type_response.status_code == 201, "创建用户类型字典失败"
|
|
user_type_id = user_type_response.json()["id"]
|
|
test_data_manager.add_dict_type(user_type_id)
|
|
|
|
# 步骤5: 添加普通用户类型
|
|
normal_type_data = {
|
|
"type": f"USER_TYPE_{unique_id}",
|
|
"code": "NORMAL",
|
|
"name": "普通用户",
|
|
"value": "1",
|
|
"remark": "普通注册用户",
|
|
"sort": 1
|
|
}
|
|
normal_type_response = await dict_api.create(normal_type_data)
|
|
assert normal_type_response.status_code == 201, "添加普通用户类型失败"
|
|
normal_type_id = normal_type_response.json()["id"]
|
|
test_data_manager.add_dict(normal_type_id)
|
|
|
|
# 步骤6: 添加管理员类型
|
|
admin_type_data = {
|
|
"type": f"USER_TYPE_{unique_id}",
|
|
"code": "ADMIN",
|
|
"name": "管理员",
|
|
"value": "2",
|
|
"remark": "系统管理员",
|
|
"sort": 2
|
|
}
|
|
admin_type_response = await dict_api.create(admin_type_data)
|
|
assert admin_type_response.status_code == 201, "添加管理员类型失败"
|
|
admin_type_id = admin_type_response.json()["id"]
|
|
test_data_manager.add_dict(admin_type_id)
|
|
|
|
# 步骤7: 修改字典数据
|
|
update_status_data = {
|
|
"name": "已启用",
|
|
"remark": "UAT测试-更新启用状态描述"
|
|
}
|
|
update_status_response = await dict_api.update_dict(enabled_status_id, update_status_data)
|
|
assert update_status_response.status_code == 200, "更新字典数据失败"
|
|
|
|
# 步骤8: 验证字典数据更新
|
|
verify_status_response = await dict_api.get_dict_by_id(enabled_status_id)
|
|
assert verify_status_response.status_code == 200, "验证字典数据失败"
|
|
verify_status_data = verify_status_response.json()
|
|
assert verify_status_data["name"] == "已启用", "字典数据更新未生效"
|
|
|
|
# 步骤9: 查询字典类型列表
|
|
type_list_response = await dict_api.get_type_list(page=0, size=10)
|
|
assert type_list_response.status_code == 200, "获取字典类型列表失败"
|
|
type_list = type_list_response.json()
|
|
assert any(t["type"] == f"USER_STATUS_{unique_id}" for t in type_list), "字典类型未创建成功"
|
|
|
|
# 步骤10: 查询字典数据列表
|
|
data_list_response = await dict_api.get_dict_list(page=0, size=10)
|
|
assert data_list_response.status_code == 200, "获取字典数据列表失败"
|
|
data_list = data_list_response.json()
|
|
assert len(data_list["content"]) >= 5, "字典数据数量不足"
|
|
|
|
# 清理
|
|
await dict_api.delete_dict(admin_type_id)
|
|
test_data_manager._dicts.remove(admin_type_id)
|
|
await dict_api.delete_dict(normal_type_id)
|
|
test_data_manager._dicts.remove(normal_type_id)
|
|
await dict_api.delete_type(user_type_id)
|
|
test_data_manager._dict_types.remove(user_type_id)
|
|
|
|
await dict_api.delete_dict(disabled_status_id)
|
|
test_data_manager._dicts.remove(disabled_status_id)
|
|
await dict_api.delete_dict(enabled_status_id)
|
|
test_data_manager._dicts.remove(enabled_status_id)
|
|
await dict_api.delete_type(status_type_id)
|
|
test_data_manager._dict_types.remove(status_type_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_notice_management(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-05: 通知公告验证
|
|
|
|
场景描述：
|
|
- 创建系统通知
|
|
- 发布通知
|
|
- 查看通知
|
|
- 更新通知
|
|
- 删除通知
|
|
"""
|
|
notice_api = SysNoticeAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_notice_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 创建系统通知
|
|
notice_data = {
|
|
"noticeTitle": f"系统维护通知_{unique_id}",
|
|
"noticeType": "1",
|
|
"noticeContent": f"""
|
|
<h1>系统维护通知</h1>
|
|
<p>尊敬的用户：</p>
|
|
<p>为提升系统性能，我们将于以下时间进行系统维护：</p>
|
|
<ul>
|
|
<li>维护时间：2024-01-01 02:00:00 至 2024-01-01 06:00:00</li>
|
|
<li>维护内容：数据库优化、服务器升级</li>
|
|
<li>影响范围：系统暂时不可用</li>
|
|
</ul>
|
|
<p>给您带来不便，敬请谅解！</p>
|
|
<p>技术支持团队</p>
|
|
""",
|
|
"status": "0",
|
|
"remark": "UAT测试-系统维护通知"
|
|
}
|
|
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:
|
|
all_notices = await notice_api.get_all()
|
|
notices = all_notices.json()
|
|
notice = next((n for n in notices if unique_id in n["noticeTitle"]), None)
|
|
notice_id = notice["id"] if notice else None
|
|
assert notice_id is not None, "获取通知ID失败"
|
|
test_data_manager.add_notice(notice_id)
|
|
|
|
# 步骤2: 验证通知创建成功
|
|
get_notice_response = await notice_api.get_by_id(notice_id)
|
|
assert get_notice_response.status_code == 200, "获取通知详情失败"
|
|
get_notice_data = get_notice_response.json()
|
|
assert unique_id in get_notice_data["noticeTitle"], "通知标题不匹配"
|
|
|
|
# 步骤3: 更新通知内容
|
|
update_notice_data = {
|
|
"noticeTitle": f"系统维护通知_已更新_{unique_id}",
|
|
"noticeContent": """
|
|
<h1>系统维护通知（更新版）</h1>
|
|
<p>尊敬的用户：</p>
|
|
<p>为提升系统性能，我们将于以下时间进行系统维护：</p>
|
|
<ul>
|
|
<li>维护时间：2024-01-01 02:00:00 至 2024-01-01 06:00:00</li>
|
|
<li>维护内容：数据库优化、服务器升级、安全补丁</li>
|
|
<li>影响范围：系统暂时不可用</li>
|
|
<li>更新说明：新增安全补丁更新</li>
|
|
</ul>
|
|
<p>给您带来不便，敬请谅解！</p>
|
|
<p>技术支持团队</p>
|
|
""",
|
|
"remark": "UAT测试-更新通知内容"
|
|
}
|
|
update_notice_response = await notice_api.update(notice_id, update_notice_data)
|
|
assert update_notice_response.status_code == 200, "更新通知失败"
|
|
|
|
# 步骤4: 验证通知更新成功
|
|
verify_notice_response = await notice_api.get_by_id(notice_id)
|
|
assert verify_notice_response.status_code == 200, "验证通知更新失败"
|
|
verify_notice_data = verify_notice_response.json()
|
|
assert "已更新" in verify_notice_data["noticeTitle"], "通知更新未生效"
|
|
assert "安全补丁" in verify_notice_data["noticeContent"], "通知内容更新未生效"
|
|
|
|
# 步骤5: 分页查询通知
|
|
notice_list_response = await notice_api.get_list(page=0, size=10)
|
|
assert notice_list_response.status_code == 200, "分页查询通知失败"
|
|
notice_list = notice_list_response.json()
|
|
assert "content" in notice_list, "通知列表缺少content字段"
|
|
assert "totalElements" in notice_list, "通知列表缺少totalElements字段"
|
|
|
|
# 步骤6: 删除通知
|
|
delete_notice_response = await notice_api.delete(notice_id)
|
|
assert delete_notice_response.status_code in [200, 204], "删除通知失败"
|
|
|
|
# 步骤7: 验证通知已删除
|
|
verify_delete_response = await notice_api.get_by_id(notice_id)
|
|
assert verify_delete_response.status_code in [404, 500], "删除的通知不应能查询到"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_file_management(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-06: 文件管理验证
|
|
|
|
场景描述：
|
|
- 上传文件
|
|
- 查看文件列表
|
|
- 下载文件
|
|
- 删除文件
|
|
- 文件分类管理
|
|
"""
|
|
file_api = FileAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_file_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 上传测试文件
|
|
test_file_content = f"""
|
|
UAT测试文件
|
|
文件ID: {unique_id}
|
|
创建时间: {time.strftime('%Y-%m-%d %H:%M:%S')}
|
|
文件内容: 这是用于UAT验收测试的文件内容
|
|
""".encode('utf-8')
|
|
|
|
upload_response = await file_api.upload_file(
|
|
f"uat_test_file_{unique_id}.txt",
|
|
test_file_content
|
|
)
|
|
assert upload_response.status_code == 200, "上传文件失败"
|
|
upload_data = upload_response.json()
|
|
file_id = upload_data.get("id") or upload_data.get("fileId")
|
|
assert file_id is not None, "获取文件ID失败"
|
|
test_data_manager.add_file(file_id)
|
|
|
|
# 步骤2: 验证文件上传成功
|
|
file_info_response = await file_api.get_file_info(file_id)
|
|
assert file_info_response.status_code == 200, "获取文件信息失败"
|
|
file_info = file_info_response.json()
|
|
assert f"uat_test_file_{unique_id}" in file_info.get("fileName", ""), "文件名不匹配"
|
|
|
|
# 步骤3: 查询文件列表
|
|
file_list_response = await file_api.get_file_list(page=0, size=10)
|
|
assert file_list_response.status_code == 200, "查询文件列表失败"
|
|
file_list = file_list_response.json()
|
|
assert "content" in file_list, "文件列表缺少content字段"
|
|
assert "totalElements" in file_list, "文件列表缺少totalElements字段"
|
|
|
|
# 步骤4: 下载文件
|
|
download_response = await file_api.download_file(file_id)
|
|
assert download_response.status_code == 200, "下载文件失败"
|
|
downloaded_content = download_response.content
|
|
assert downloaded_content == test_file_content, "下载的文件内容不匹配"
|
|
|
|
# 步骤5: 更新文件信息
|
|
update_file_data = {
|
|
"fileName": f"uat_test_file_{unique_id}_updated",
|
|
"fileType": "txt",
|
|
"remark": "UAT测试-更新文件信息"
|
|
}
|
|
update_file_response = await file_api.update_file(file_id, update_file_data)
|
|
assert update_file_response.status_code == 200, "更新文件信息失败"
|
|
|
|
# 步骤6: 删除文件
|
|
delete_file_response = await file_api.delete_file(file_id)
|
|
assert delete_file_response.status_code in [200, 204], "删除文件失败"
|
|
|
|
# 步骤7: 验证文件已删除
|
|
verify_delete_response = await file_api.get_file_info(file_id)
|
|
assert verify_delete_response.status_code in [404, 500], "删除的文件不应能查询到"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_audit_log_verification(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-07: 审计日志验证
|
|
|
|
场景描述：
|
|
- 执行操作生成审计日志
|
|
- 查询操作日志
|
|
- 查询登录日志
|
|
- 查询异常日志
|
|
- 日志筛选和分页
|
|
"""
|
|
user_api = UserAPI(authenticated_client)
|
|
audit_api = AuditAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_audit_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 创建测试用户以触发审计日志
|
|
audit_user_data = {
|
|
"username": f"audit_user_{unique_id}",
|
|
"password": "Audit123!@#",
|
|
"email": f"audit_{unique_id}@example.com",
|
|
"status": 1,
|
|
"remark": "UAT测试-审计日志用户"
|
|
}
|
|
audit_user_response = await user_api.create_user(audit_user_data)
|
|
assert audit_user_response.status_code == 201, "创建审计日志测试用户失败"
|
|
audit_user_id = audit_user_response.json()["id"]
|
|
test_data_manager.add_user(audit_user_id)
|
|
|
|
# 步骤2: 等待审计日志生成
|
|
await asyncio.sleep(1)
|
|
|
|
# 步骤3: 查询操作日志
|
|
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, "查询操作日志失败"
|
|
operation_logs = operation_log_response.json()
|
|
assert "content" in operation_logs, "操作日志缺少content字段"
|
|
assert "totalElements" in operation_logs, "操作日志缺少totalElements字段"
|
|
|
|
# 步骤4: 查询登录日志
|
|
login_log_response = await audit_api.get_login_logs(page=0, size=10)
|
|
assert login_log_response.status_code == 200, "查询登录日志失败"
|
|
login_logs = login_log_response.json()
|
|
assert "content" in login_logs, "登录日志缺少content字段"
|
|
assert "totalElements" in login_logs, "登录日志缺少totalElements字段"
|
|
|
|
# 步骤5: 查询异常日志
|
|
exception_log_response = await audit_api.get_exception_logs(page=0, size=10)
|
|
assert exception_log_response.status_code == 200, "查询异常日志失败"
|
|
exception_logs = exception_log_response.json()
|
|
assert "content" in exception_logs, "异常日志缺少content字段"
|
|
assert "totalElements" in exception_logs, "异常日志缺少totalElements字段"
|
|
|
|
# 步骤6: 按时间范围筛选日志
|
|
current_time = int(time.time() * 1000)
|
|
time_range_response = await audit_api.get_operation_logs(
|
|
page=0, size=10,
|
|
beginTime=str(current_time - 3600000),
|
|
endTime=str(current_time + 3600000)
|
|
)
|
|
assert time_range_response.status_code == 200, "按时间范围筛选日志失败"
|
|
|
|
# 步骤7: 按用户筛选日志
|
|
user_filter_response = await audit_api.get_operation_logs(
|
|
page=0, size=10,
|
|
userName=f"audit_user_{unique_id}"
|
|
)
|
|
assert user_filter_response.status_code == 200, "按用户筛选日志失败"
|
|
|
|
# 步骤8: 清理测试用户
|
|
await user_api.delete_user(audit_user_id)
|
|
test_data_manager._users.remove(audit_user_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_performance_acceptance(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-08: 性能验收验证
|
|
|
|
场景描述：
|
|
- 测试接口响应时间
|
|
- 测试并发用户处理能力
|
|
- 测试大数据量查询性能
|
|
- 验证性能指标达标
|
|
"""
|
|
user_api = UserAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_perf_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
performance_metrics = {}
|
|
|
|
# 步骤1: 创建测试用户
|
|
perf_user_data = {
|
|
"username": f"perf_user_{unique_id}",
|
|
"password": "Perf123!@#",
|
|
"email": f"perf_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
perf_user_response = await user_api.create_user(perf_user_data)
|
|
assert perf_user_response.status_code == 201, "创建性能测试用户失败"
|
|
perf_user_id = perf_user_response.json()["id"]
|
|
test_data_manager.add_user(perf_user_id)
|
|
|
|
# 步骤2: 测试单用户查询性能
|
|
start_time = time.time()
|
|
user_query_response = await user_api.get_user_by_id(perf_user_id)
|
|
query_duration = (time.time() - start_time) * 1000
|
|
assert user_query_response.status_code == 200, "查询用户失败"
|
|
assert query_duration < 500, f"单用户查询性能不达标: {query_duration}ms"
|
|
performance_metrics["single_user_query_ms"] = query_duration
|
|
|
|
# 步骤3: 创建多个用户用于并发测试
|
|
concurrent_user_ids = []
|
|
for i in range(10):
|
|
concurrent_user_data = {
|
|
"username": f"perf_concurrent_{unique_id}_{i}",
|
|
"password": "Perf123!@#",
|
|
"email": f"perf_concurrent_{unique_id}_{i}@example.com",
|
|
"status": 1
|
|
}
|
|
concurrent_user_response = await user_api.create_user(concurrent_user_data)
|
|
assert concurrent_user_response.status_code == 201, f"创建并发测试用户{i}失败"
|
|
concurrent_user_ids.append(concurrent_user_response.json()["id"])
|
|
test_data_manager.add_user(concurrent_user_ids[-1])
|
|
|
|
# 步骤4: 测试并发查询性能（5个并发）
|
|
start_time = time.time()
|
|
|
|
async def query_user(user_id):
|
|
response = await user_api.get_user_by_id(user_id)
|
|
return response.status_code == 200
|
|
|
|
tasks = [query_user(user_id) for user_id in concurrent_user_ids[:5]]
|
|
results = await asyncio.gather(*tasks)
|
|
|
|
concurrent_query_duration = (time.time() - start_time) * 1000
|
|
assert all(results), "并发查询存在失败"
|
|
assert concurrent_query_duration < 2000, f"并发查询性能不达标: {concurrent_query_duration}ms"
|
|
performance_metrics["concurrent_query_5_users_ms"] = concurrent_query_duration
|
|
|
|
# 步骤5: 测试分页查询性能
|
|
pagination_start = time.time()
|
|
pagination_response = await user_api.get_users_by_page(page=0, size=100)
|
|
pagination_duration = (time.time() - pagination_start) * 1000
|
|
assert pagination_response.status_code == 200, "分页查询失败"
|
|
assert pagination_duration < 1000, f"分页查询性能不达标: {pagination_duration}ms"
|
|
performance_metrics["pagination_query_100_ms"] = pagination_duration
|
|
|
|
# 步骤6: 验证性能指标
|
|
assert performance_metrics["single_user_query_ms"] < 500, "单用户查询性能不达标"
|
|
assert performance_metrics["concurrent_query_5_users_ms"] < 2000, "并发查询性能不达标"
|
|
assert performance_metrics["pagination_query_100_ms"] < 1000, "分页查询性能不达标"
|
|
|
|
# 步骤7: 清理测试用户
|
|
for user_id in concurrent_user_ids:
|
|
await user_api.delete_user(user_id)
|
|
test_data_manager._users.remove(user_id)
|
|
await user_api.delete_user(perf_user_id)
|
|
test_data_manager._users.remove(perf_user_id)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_security_acceptance(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-09: 安全验收验证
|
|
|
|
场景描述：
|
|
- 测试SQL注入防护
|
|
- 测试XSS攻击防护
|
|
- 测试认证授权
|
|
- 测试密码策略
|
|
- 测试敏感信息保护
|
|
"""
|
|
user_api = UserAPI(authenticated_client)
|
|
|
|
unique_id = f"uat_security_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# 步骤1: 测试SQL注入防护
|
|
sql_injection_username = f"test' OR '1'='1' -- {unique_id}"
|
|
sql_injection_data = {
|
|
"username": sql_injection_username,
|
|
"password": "Test123!@#",
|
|
"email": f"sql_injection_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
sql_injection_response = await user_api.create_user(sql_injection_data)
|
|
# 应该成功创建（转义处理），但不应被SQL注入影响
|
|
assert sql_injection_response.status_code in [201, 400, 422], "SQL注入尝试应被拦截或正确处理"
|
|
|
|
# 步骤2: 测试XSS攻击防护
|
|
xss_username = f"<script>alert('XSS')</script>_{unique_id}"
|
|
xss_data = {
|
|
"username": xss_username,
|
|
"password": "Test123!@#",
|
|
"email": f"xss_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
xss_response = await user_api.create_user(xss_data)
|
|
# 应该被正确处理（转义或拒绝）
|
|
assert xss_response.status_code in [201, 400, 422], "XSS攻击应被拦截或正确处理"
|
|
|
|
# 步骤3: 测试弱密码检测
|
|
weak_password_data = {
|
|
"username": f"weak_password_{unique_id}",
|
|
"password": "123456",
|
|
"email": f"weak_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
weak_password_response = await user_api.create_user(weak_password_data)
|
|
# 应该被拒绝
|
|
assert weak_password_response.status_code in [400, 422], "弱密码应被拒绝"
|
|
|
|
# 步骤4: 测试重复用户名检测
|
|
duplicate_username = f"duplicate_{unique_id}"
|
|
duplicate_data1 = {
|
|
"username": duplicate_username,
|
|
"password": "Test123!@#",
|
|
"email": f"duplicate1_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
duplicate_data2 = {
|
|
"username": duplicate_username,
|
|
"password": "Test123!@#",
|
|
"email": f"duplicate2_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
duplicate_response1 = await user_api.create_user(duplicate_data1)
|
|
assert duplicate_response1.status_code == 201, "创建第一个重复用户名用户失败"
|
|
duplicate_response2 = await user_api.create_user(duplicate_data2)
|
|
assert duplicate_response2.status_code in [400, 409, 422], "重复用户名应被拒绝"
|
|
|
|
# 步骤5: 测试无效邮箱格式
|
|
invalid_email_data = {
|
|
"username": f"invalid_email_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": "invalid-email-format",
|
|
"status": 1
|
|
}
|
|
invalid_email_response = await user_api.create_user(invalid_email_data)
|
|
assert invalid_email_response.status_code in [400, 422], "无效邮箱格式应被拒绝"
|
|
|
|
# 步骤6: 测试空用户名
|
|
empty_username_data = {
|
|
"username": "",
|
|
"password": "Test123!@#",
|
|
"email": f"empty_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
empty_username_response = await user_api.create_user(empty_username_data)
|
|
assert empty_username_response.status_code in [400, 422], "空用户名应被拒绝"
|
|
|
|
# 步骤7: 测试空密码
|
|
empty_password_data = {
|
|
"username": f"empty_password_{unique_id}",
|
|
"password": "",
|
|
"email": f"empty_password_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
empty_password_response = await user_api.create_user(empty_password_data)
|
|
assert empty_password_response.status_code in [400, 422], "空密码应被拒绝"
|
|
|
|
# 步骤8: 测试超长输入
|
|
long_username = "a" * 200
|
|
long_data = {
|
|
"username": long_username,
|
|
"password": "Test123!@#",
|
|
"email": f"long_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
long_response = await user_api.create_user(long_data)
|
|
assert long_response.status_code in [400, 422], "超长输入应被拒绝"
|
|
|
|
# 步骤9: 测试特殊字符处理
|
|
special_char_data = {
|
|
"username": f"special!@#$%^&*()_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"special_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
special_char_response = await user_api.create_user(special_char_data)
|
|
# 特殊字符应被正确处理
|
|
assert special_char_response.status_code in [201, 400, 422], "特殊字符应被正确处理"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uat_business_workflow(
|
|
self, authenticated_client, test_data_manager
|
|
):
|
|
"""
|
|
UAT-10: 完整业务流程验证
|
|
|
|
场景描述：
|
|
- 从用户创建到权限分配的完整流程
|
|
- 从角色配置到菜单授权的完整流程
|
|
- 从数据字典维护到配置管理的完整流程
|
|
- 验证端到端业务流程
|
|
"""
|
|
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"uat_workflow_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
|
|
|
|
# ==================== 第一部分：用户与角色 ====================
|
|
# 1.1 创建部门经理角色
|
|
manager_role_data = {
|
|
"roleName": f"Workflow_Manager_Role_{unique_id}",
|
|
"roleKey": f"workflow_manager_role_{unique_id}",
|
|
"roleSort": 5,
|
|
"status": 1,
|
|
"remark": "UAT测试-业务流程-部门经理"
|
|
}
|
|
manager_role_response = await role_api.create_role(manager_role_data)
|
|
assert manager_role_response.status_code == 201, "创建部门经理角色失败"
|
|
manager_role_id = manager_role_response.json()["id"]
|
|
test_data_manager.add_role(manager_role_id)
|
|
|
|
# 1.2 创建普通员工角色
|
|
employee_role_data = {
|
|
"roleName": f"Workflow_Employee_Role_{unique_id}",
|
|
"roleKey": f"workflow_employee_role_{unique_id}",
|
|
"roleSort": 10,
|
|
"status": 1,
|
|
"remark": "UAT测试-业务流程-普通员工"
|
|
}
|
|
employee_role_response = await role_api.create_role(employee_role_data)
|
|
assert employee_role_response.status_code == 201, "创建普通员工角色失败"
|
|
employee_role_id = employee_role_response.json()["id"]
|
|
test_data_manager.add_role(employee_role_id)
|
|
|
|
# 1.3 创建系统管理员角色
|
|
admin_role_data = {
|
|
"roleName": f"Workflow_Admin_Role_{unique_id}",
|
|
"roleKey": f"workflow_admin_role_{unique_id}",
|
|
"roleSort": 1,
|
|
"status": 1,
|
|
"remark": "UAT测试-业务流程-系统管理员"
|
|
}
|
|
admin_role_response = await role_api.create_role(admin_role_data)
|
|
assert admin_role_response.status_code == 201, "创建系统管理员角色失败"
|
|
admin_role_id = admin_role_response.json()["id"]
|
|
test_data_manager.add_role(admin_role_id)
|
|
|
|
# ==================== 第二部分：菜单与权限 ====================
|
|
# 2.1 创建系统管理主菜单
|
|
system_menu_data = {
|
|
"parentId": 0,
|
|
"menuName": f"系统管理_{unique_id}",
|
|
"menuType": "M",
|
|
"orderNum": 1,
|
|
"component": "Layout",
|
|
"perms": f"system:workflow:{unique_id}",
|
|
"status": 1,
|
|
"icon": "system"
|
|
}
|
|
system_menu_response = await menu_api.create_menu(system_menu_data)
|
|
assert system_menu_response.status_code == 201, "创建系统管理菜单失败"
|
|
system_menu_id = system_menu_response.json()["id"]
|
|
test_data_manager.add_menu(system_menu_id)
|
|
|
|
# 2.2 创建用户管理子菜单
|
|
user_menu_data = {
|
|
"parentId": system_menu_id,
|
|
"menuName": "用户管理",
|
|
"menuType": "C",
|
|
"orderNum": 1,
|
|
"component": "UserManagement",
|
|
"perms": f"user:workflow:{unique_id}",
|
|
"status": 1
|
|
}
|
|
user_menu_response = await menu_api.create_menu(user_menu_data)
|
|
assert user_menu_response.status_code == 201, "创建用户管理菜单失败"
|
|
user_menu_id = user_menu_response.json()["id"]
|
|
test_data_manager.add_menu(user_menu_id)
|
|
|
|
# 2.3 创建角色管理子菜单
|
|
role_menu_data = {
|
|
"parentId": system_menu_id,
|
|
"menuName": "角色管理",
|
|
"menuType": "C",
|
|
"orderNum": 2,
|
|
"component": "RoleManagement",
|
|
"perms": f"role:workflow:{unique_id}",
|
|
"status": 1
|
|
}
|
|
role_menu_response = await menu_api.create_menu(role_menu_data)
|
|
assert role_menu_response.status_code == 201, "创建角色管理菜单失败"
|
|
role_menu_id = role_menu_response.json()["id"]
|
|
test_data_manager.add_menu(role_menu_id)
|
|
|
|
# 2.4 创建权限分配
|
|
admin_permission_data = {"menuIds": [system_menu_id, user_menu_id, role_menu_id]}
|
|
admin_permission_response = await role_api.assign_permissions(admin_role_id, admin_permission_data)
|
|
assert admin_permission_response.status_code == 200, "为管理员分配权限失败"
|
|
|
|
manager_permission_data = {"menuIds": [user_menu_id]}
|
|
manager_permission_response = await role_api.assign_permissions(manager_role_id, manager_permission_data)
|
|
assert manager_permission_response.status_code == 200, "为经理分配权限失败"
|
|
|
|
employee_permission_data = {"menuIds": []}
|
|
employee_permission_response = await role_api.assign_permissions(employee_role_id, employee_permission_data)
|
|
assert employee_permission_response.status_code == 200, "为员工分配权限失败"
|
|
|
|
# ==================== 第三部分：数据字典 ====================
|
|
# 3.1 创建用户状态字典
|
|
status_type_data = {
|
|
"type": f"WORKFLOW_STATUS_{unique_id}",
|
|
"name": "用户状态",
|
|
"remark": "UAT测试-业务流程-用户状态",
|
|
"sort": 1
|
|
}
|
|
status_type_response = await dict_api.create_type(status_type_data)
|
|
assert status_type_response.status_code == 201, "创建用户状态字典类型失败"
|
|
status_type_id = status_type_response.json()["id"]
|
|
test_data_manager.add_dict_type(status_type_id)
|
|
|
|
# 3.2 添加状态数据
|
|
enabled_status_data = {
|
|
"type": f"WORKFLOW_STATUS_{unique_id}",
|
|
"code": "ENABLED",
|
|
"name": "启用",
|
|
"value": "1",
|
|
"remark": "用户已启用",
|
|
"sort": 1
|
|
}
|
|
enabled_status_response = await dict_api.create(enabled_status_data)
|
|
assert enabled_status_response.status_code == 201, "添加启用状态失败"
|
|
enabled_status_id = enabled_status_response.json()["id"]
|
|
test_data_manager.add_dict(enabled_status_id)
|
|
|
|
disabled_status_data = {
|
|
"type": f"WORKFLOW_STATUS_{unique_id}",
|
|
"code": "DISABLED",
|
|
"name": "禁用",
|
|
"value": "0",
|
|
"remark": "用户已禁用",
|
|
"sort": 2
|
|
}
|
|
disabled_status_response = await dict_api.create(disabled_status_data)
|
|
assert disabled_status_response.status_code == 201, "添加禁用状态失败"
|
|
disabled_status_id = disabled_status_response.json()["id"]
|
|
test_data_manager.add_dict(disabled_status_id)
|
|
|
|
# ==================== 第四部分：系统配置 ====================
|
|
# 4.1 创建系统配置
|
|
system_config_data = {
|
|
"configKey": f"workflow_config_{unique_id}",
|
|
"configName": "UAT业务流程配置",
|
|
"configType": "Y",
|
|
"configValue": '{"maxUsers":1000,"enableAudit":true}',
|
|
"remark": "UAT测试-业务流程-系统配置"
|
|
}
|
|
system_config_response = await config_api.create_config(system_config_data)
|
|
assert system_config_response.status_code == 201, "创建系统配置失败"
|
|
system_config_id = system_config_response.json()["id"]
|
|
test_data_manager.add_config(system_config_id)
|
|
|
|
# ==================== 第五部分：用户创建与管理 ====================
|
|
# 5.1 创建测试用户
|
|
test_user_data = {
|
|
"username": f"workflow_user_{unique_id}",
|
|
"password": "Workflow123!@#",
|
|
"email": f"workflow_{unique_id}@company.com",
|
|
"roleId": employee_role_id,
|
|
"status": 1,
|
|
"remark": "UAT测试-业务流程-测试用户"
|
|
}
|
|
test_user_response = await user_api.create_user(test_user_data)
|
|
assert test_user_response.status_code == 201, "创建测试用户失败"
|
|
test_user_id = test_user_response.json()["id"]
|
|
test_data_manager.add_user(test_user_id)
|
|
|
|
# 5.2 验证用户创建成功
|
|
user_detail_response = await user_api.get_user_by_id(test_user_id)
|
|
assert user_detail_response.status_code == 200, "验证用户详情失败"
|
|
user_detail = user_detail_response.json()
|
|
assert user_detail["username"] == f"workflow_user_{unique_id}", "用户详情不匹配"
|
|
|
|
# 5.3 更新用户角色（晋升）
|
|
update_user_role_data = {"roleId": manager_role_id}
|
|
update_user_role_response = await user_api.update_user(test_user_id, update_user_role_data)
|
|
assert update_user_role_response.status_code == 200, "更新用户角色失败"
|
|
|
|
# 5.4 验证角色更新
|
|
verify_role_response = await user_api.get_user_by_id(test_user_id)
|
|
assert verify_role_response.status_code == 200, "验证角色更新失败"
|
|
verify_role_detail = verify_role_response.json()
|
|
assert verify_role_detail["roleId"] == manager_role_id, "角色更新未生效"
|
|
|
|
# 5.5 更新用户个人信息
|
|
update_profile_data = {
|
|
"nickName": f"测试用户_{unique_id}",
|
|
"phone": f"1380013800{unique_id[-4:]}",
|
|
"remark": "UAT测试-业务流程-更新个人信息"
|
|
}
|
|
update_profile_response = await user_api.update_user_profile(update_profile_data)
|
|
assert update_profile_response.status_code == 200, "更新个人信息失败"
|
|
|
|
# ==================== 第六部分：权限验证 ====================
|
|
# 6.1 验证管理员权限
|
|
admin_menus_response = await menu_api.get_user_menus_by_role(admin_role_id)
|
|
assert admin_menus_response.status_code == 200, "获取管理员菜单失败"
|
|
admin_menus = admin_menus_response.json()
|
|
assert len(admin_menus) >= 3, "管理员应能访问至少3个菜单"
|
|
|
|
# 6.2 验证经理权限
|
|
manager_menus_response = await menu_api.get_user_menus_by_role(manager_role_id)
|
|
assert manager_menus_response.status_code == 200, "获取经理菜单失败"
|
|
manager_menus = manager_menus_response.json()
|
|
assert len(manager_menus) == 1, "经理应只能访问1个菜单"
|
|
assert manager_menus[0]["id"] == user_menu_id, "经理菜单权限不正确"
|
|
|
|
# 6.3 验证员工权限
|
|
employee_menus_response = await menu_api.get_user_menus_by_role(employee_role_id)
|
|
assert employee_menus_response.status_code == 200, "获取员工菜单失败"
|
|
employee_menus = employee_menus_response.json()
|
|
assert len(employee_menus) == 0, "员工应无菜单权限"
|
|
|
|
# ==================== 第七部分：清理 ====================
|
|
# 7.1 删除测试用户
|
|
await user_api.delete_user(test_user_id)
|
|
test_data_manager._users.remove(test_user_id)
|
|
|
|
# 7.2 清理权限分配
|
|
await role_api.assign_permissions(admin_role_id, {"menuIds": []})
|
|
await role_api.assign_permissions(manager_role_id, {"menuIds": []})
|
|
await role_api.assign_permissions(employee_role_id, {"menuIds": []})
|
|
|
|
# 7.3 删除菜单
|
|
await menu_api.delete_menu(role_menu_id)
|
|
test_data_manager._menus.remove(role_menu_id)
|
|
await menu_api.delete_menu(user_menu_id)
|
|
test_data_manager._menus.remove(user_menu_id)
|
|
await menu_api.delete_menu(system_menu_id)
|
|
test_data_manager._menus.remove(system_menu_id)
|
|
|
|
# 7.4 删除角色
|
|
await role_api.delete_role(admin_role_id)
|
|
test_data_manager._roles.remove(admin_role_id)
|
|
await role_api.delete_role(manager_role_id)
|
|
test_data_manager._roles.remove(manager_role_id)
|
|
await role_api.delete_role(employee_role_id)
|
|
test_data_manager._roles.remove(employee_role_id)
|
|
|
|
# 7.5 删除字典数据
|
|
await dict_api.delete_dict(disabled_status_id)
|
|
test_data_manager._dicts.remove(disabled_status_id)
|
|
await dict_api.delete_dict(enabled_status_id)
|
|
test_data_manager._dicts.remove(enabled_status_id)
|
|
await dict_api.delete_type(status_type_id)
|
|
test_data_manager._dict_types.remove(status_type_id)
|
|
|
|
# 7.6 删除系统配置
|
|
await config_api.delete_config(system_config_id)
|
|
test_data_manager._configs.remove(system_config_id)
|