张翔
1e3dc11d59
feat(test-suite): 新增测试套件模块，包含API测试客户端和测试配置 fix(api): 修复数据库实体和仓库的删除操作返回值 style(api): 统一数据库表名和字段命名 perf(api): 添加缓存注解提升配置查询性能 test(api): 添加H2测试数据库配置支持 chore: 清理旧的测试文件和脚本
160 lines
5.7 KiB
Python
160 lines
5.7 KiB
Python
"""
|
|
数据恢复和备份测试用例
|
|
测试数据备份、恢复和完整性验证
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
import time
|
|
from api.user_api import UserAPI
|
|
from api.role_api import RoleAPI
|
|
from api.notice_api import SysNoticeAPI
|
|
|
|
|
|
@pytest.mark.recovery
|
|
@pytest.mark.regression
|
|
@pytest.mark.critical
|
|
class TestDataRecovery:
|
|
"""数据恢复和备份测试类"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_user_data_backup_and_restore(self, authenticated_client, test_data_manager):
|
|
"""测试用户数据备份和恢复"""
|
|
user_api = UserAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}"
|
|
|
|
# 创建测试用户
|
|
user_data = {
|
|
"username": f"backup_user_{unique_id}",
|
|
"password": "Test123!@#",
|
|
"email": f"backup_{unique_id}@example.com",
|
|
"status": 1
|
|
}
|
|
|
|
create_response = await user_api.create_user(user_data)
|
|
assert create_response.status_code == 201
|
|
user_id = create_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# 备份用户数据（模拟备份操作）
|
|
backup_data = create_response.json()
|
|
|
|
# 修改用户数据
|
|
update_data = {"email": f"updated_{unique_id}@example.com"}
|
|
await user_api.update_user(user_id, update_data)
|
|
|
|
# 验证数据已修改
|
|
updated_user = await user_api.get_user_by_id(user_id)
|
|
assert updated_user.json()["email"] == update_data["email"]
|
|
|
|
# 恢复数据（模拟恢复操作）
|
|
restore_response = await user_api.update_user(user_id, {
|
|
"email": backup_data["email"],
|
|
"username": backup_data["username"]
|
|
})
|
|
assert restore_response.status_code == 200
|
|
|
|
# 验证数据已恢复
|
|
restored_user = await user_api.get_user_by_id(user_id)
|
|
assert restored_user.json()["email"] == backup_data["email"]
|
|
assert restored_user.json()["username"] == backup_data["username"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_role_data_backup_and_restore(self, authenticated_client, test_data_manager):
|
|
"""测试角色数据备份和恢复"""
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}"
|
|
|
|
# 创建测试角色
|
|
role_data = {
|
|
"roleName": f"Backup_Role_{unique_id}",
|
|
"roleKey": f"backup_role_{unique_id}",
|
|
"roleSort": 1,
|
|
"status": 1
|
|
}
|
|
|
|
create_response = await role_api.create_role(role_data)
|
|
assert create_response.status_code == 201
|
|
role_id = create_response.json()["id"]
|
|
test_data_manager.add_role(role_id)
|
|
|
|
# 备份角色数据
|
|
backup_data = create_response.json()
|
|
|
|
# 修改角色数据
|
|
update_data = {"roleName": f"Updated_Role_{unique_id}"}
|
|
await role_api.update_role(role_id, update_data)
|
|
|
|
# 验证数据已修改
|
|
updated_role = await role_api.get_role_by_id(role_id)
|
|
assert updated_role.json()["roleName"] == update_data["roleName"]
|
|
|
|
# 恢复数据
|
|
restore_response = await role_api.update_role(role_id, {
|
|
"roleName": backup_data["roleName"],
|
|
"roleKey": backup_data["roleKey"]
|
|
})
|
|
assert restore_response.status_code == 200
|
|
|
|
# 验证数据已恢复
|
|
restored_role = await role_api.get_role_by_id(role_id)
|
|
assert restored_role.json()["roleName"] == backup_data["roleName"]
|
|
assert restored_role.json()["roleKey"] == backup_data["roleKey"]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_data_integrity_after_restore(self, authenticated_client, test_data_manager):
|
|
"""测试恢复后数据完整性"""
|
|
user_api = UserAPI(authenticated_client)
|
|
role_api = RoleAPI(authenticated_client)
|
|
|
|
unique_id = f"{int(time.time() * 1000)}"
|
|
|
|
# 创建角色
|
|
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)
|
|
role_id = role_response.json()["id"]
|
|
test_data_manager.add_role(role_id)
|
|
|
|
# 创建用户并分配角色
|
|
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)
|
|
user_id = user_response.json()["id"]
|
|
test_data_manager.add_user(user_id)
|
|
|
|
# 备份数据
|
|
user_backup = user_response.json()
|
|
role_backup = role_response.json()
|
|
|
|
# 修改用户数据
|
|
await user_api.update_user(user_id, {"email": f"modified_{unique_id}@example.com"})
|
|
|
|
# 恢复用户数据
|
|
await user_api.update_user(user_id, {
|
|
"email": user_backup["email"],
|
|
"username": user_backup["username"]
|
|
})
|
|
|
|
# 验证完整性
|
|
restored_user = await user_api.get_user_by_id(user_id)
|
|
user_data = restored_user.json()
|
|
assert user_data["email"] == user_backup["email"]
|
|
# 验证用户仍然关联到角色（如果API返回roleId）
|
|
if "roleId" in user_data and user_data["roleId"]:
|
|
assert user_data["roleId"] == role_id
|
|
|
|
# 验证角色仍然存在
|
|
role_verify = await role_api.get_role_by_id(role_id)
|
|
assert role_verify.status_code == 200 |