novalon-manage-system/test-suite/tests/uat/test_uat_business_scenario.py

"""
UAT业务场景验收测试

测试范围:
1. 新员工入职流程
2. 员工离职流程
3. 权限变更流程
4. 组织架构调整流程
5. 系统配置变更流程
"""

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.config_api import ConfigAPI
from api.audit_api import AuditAPI
from config.settings import settings


@pytest.mark.uat
@pytest.mark.business_scenario
@pytest.mark.asyncio
class TestBusinessScenarioUAT:
    """业务场景验收测试类"""
    
    @pytest.fixture
    async def authenticated_client(self):
        """已认证的HTTP客户端"""
        async with AuthAPI.create_client() as client:
            auth_api = AuthAPI(client)
            await auth_api.login(settings.TEST_USERNAME, settings.TEST_PASSWORD)
            yield client
    
    @pytest.fixture
    async def test_data_manager(self):
        """测试数据管理器"""
        from utils.test_data_manager import TestDataManager
        manager = TestDataManager()
        yield manager
        await manager.cleanup_all()
    
    async def test_bs_new_employee_onboarding(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-01: 新员工入职流程
        
        业务场景:
        1. HR创建新员工账户
        2. 分配基础角色(普通员工)
        3. 员工首次登录并修改密码
        4. 员工完善个人信息
        5. 验证权限范围
        
        验收标准:
        - 账户创建成功
        - 角色分配正确
        - 首次登录强制修改密码
        - 个人信息可更新
        - 权限范围符合预期
        """
        user_api = UserAPI(authenticated_client)
        role_api = RoleAPI(authenticated_client)
        auth_api = AuthAPI(authenticated_client)
        
        unique_id = f"onboard_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        
        employee_role_data = {
            "roleName": f"新员工角色_{unique_id}",
            "roleKey": f"employee_role_{unique_id}",
            "roleSort": 10,
            "status": 1,
            "remark": "业务场景测试-新员工角色"
        }
        role_response = await role_api.create_role(employee_role_data)
        assert role_response.status_code == 201, "创建员工角色失败"
        role_id = role_response.json()["id"]
        test_data_manager.add_role(role_id)
        
        employee_data = {
            "username": f"new_employee_{unique_id}",
            "password": "Welcome123!@#",
            "email": f"new_employee_{unique_id}@company.com",
            "phone": f"138001380{unique_id[-4:]}",
            "roleId": role_id,
            "status": 1,
            "remark": "业务场景测试-新入职员工"
        }
        user_response = await user_api.create_user(employee_data)
        assert user_response.status_code == 201, "创建员工账户失败"
        user_id = user_response.json()["id"]
        test_data_manager.add_user(user_id)
        
        login_response = await auth_api.login(
            f"new_employee_{unique_id}",
            "Welcome123!@#"
        )
        assert login_response.status_code == 200, "员工登录失败"
        login_data = login_response.json()
        assert "token" in login_data, "登录应返回token"
        
        profile_data = {
            "nickName": f"张三_{unique_id}",
            "phone": f"139001390{unique_id[-4:]}",
            "remark": "业务场景测试-完善个人信息"
        }
        profile_response = await user_api.update_user_profile(profile_data)
        assert profile_response.status_code == 200, "更新个人信息失败"
        
        permissions_response = await role_api.get_role_permissions(role_id)
        assert permissions_response.status_code == 200, "获取权限失败"
        
        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)
    
    async def test_bs_employee_resignation(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-02: 员工离职流程
        
        业务场景:
        1. 员工提交离职申请
        2. 管理员禁用员工账户
        3. 回收员工权限
        4. 归档员工数据
        5. 验证账户无法登录
        
        验收标准:
        - 账户状态变更为禁用
        - 权限已回收
        - 数据已归档
        - 无法登录系统
        """
        user_api = UserAPI(authenticated_client)
        role_api = RoleAPI(authenticated_client)
        auth_api = AuthAPI(authenticated_client)
        
        unique_id = f"resign_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        
        role_data = {
            "roleName": f"离职测试角色_{unique_id}",
            "roleKey": f"resign_role_{unique_id}",
            "roleSort": 10,
            "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)
        
        user_data = {
            "username": f"resign_employee_{unique_id}",
            "password": "Test123!@#",
            "email": f"resign_{unique_id}@company.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)
        
        disable_data = {"status": 0}
        disable_response = await user_api.update_user(user_id, disable_data)
        assert disable_response.status_code == 200, "禁用账户失败"
        
        user_detail = await user_api.get_user_by_id(user_id)
        assert user_detail.status_code == 200
        user_info = user_detail.json()
        assert user_info["status"] == 0, "账户状态应为禁用"
        
        login_response = await auth_api.login(
            f"resign_employee_{unique_id}",
            "Test123!@#"
        )
        assert login_response.status_code != 200, "已禁用账户不应能登录"
        
        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)
    
    async def test_bs_permission_change(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-03: 权限变更流程
        
        业务场景:
        1. 员工晋升为经理
        2. 更新角色权限
        3. 验证新权限即时生效
        4. 验证旧权限已撤销
        
        验收标准:
        - 角色变更成功
        - 新权限立即生效
        - 旧权限已撤销
        - 审计日志记录完整
        """
        user_api = UserAPI(authenticated_client)
        role_api = RoleAPI(authenticated_client)
        menu_api = MenuAPI(authenticated_client)
        
        unique_id = f"perm_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        
        employee_role_data = {
            "roleName": f"员工角色_{unique_id}",
            "roleKey": f"emp_role_{unique_id}",
            "roleSort": 10,
            "status": 1
        }
        emp_role_response = await role_api.create_role(employee_role_data)
        assert emp_role_response.status_code == 201
        emp_role_id = emp_role_response.json()["id"]
        test_data_manager.add_role(emp_role_id)
        
        manager_role_data = {
            "roleName": f"经理角色_{unique_id}",
            "roleKey": f"mgr_role_{unique_id}",
            "roleSort": 5,
            "status": 1
        }
        mgr_role_response = await role_api.create_role(manager_role_data)
        assert mgr_role_response.status_code == 201
        mgr_role_id = mgr_role_response.json()["id"]
        test_data_manager.add_role(mgr_role_id)
        
        user_data = {
            "username": f"perm_user_{unique_id}",
            "password": "Test123!@#",
            "email": f"perm_{unique_id}@company.com",
            "roleId": emp_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)
        
        user_before = await user_api.get_user_by_id(user_id)
        assert user_before.json()["roleId"] == emp_role_id
        
        update_data = {"roleId": mgr_role_id}
        update_response = await user_api.update_user(user_id, update_data)
        assert update_response.status_code == 200, "角色变更失败"
        
        user_after = await user_api.get_user_by_id(user_id)
        assert user_after.json()["roleId"] == mgr_role_id, "角色变更未生效"
        
        await user_api.delete_user(user_id)
        test_data_manager._users.remove(user_id)
        await role_api.delete_role(mgr_role_id)
        test_data_manager._roles.remove(mgr_role_id)
        await role_api.delete_role(emp_role_id)
        test_data_manager._roles.remove(emp_role_id)
    
    async def test_bs_organization_restructure(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-04: 组织架构调整流程
        
        业务场景:
        1. 创建新部门
        2. 批量调整员工部门
        3. 调整部门权限
        4. 验证组织架构变更
        
        验收标准:
        - 部门创建成功
        - 员工部门调整成功
        - 权限调整成功
        - 组织架构更新正确
        """
        user_api = UserAPI(authenticated_client)
        role_api = RoleAPI(authenticated_client)
        
        unique_id = f"org_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        
        dept_role_data = {
            "roleName": f"新部门角色_{unique_id}",
            "roleKey": f"new_dept_role_{unique_id}",
            "roleSort": 8,
            "status": 1,
            "remark": "业务场景测试-新部门"
        }
        role_response = await role_api.create_role(dept_role_data)
        assert role_response.status_code == 201
        role_id = role_response.json()["id"]
        test_data_manager.add_role(role_id)
        
        users_to_create = []
        for i in range(3):
            user_data = {
                "username": f"org_user_{i}_{unique_id}",
                "password": "Test123!@#",
                "email": f"org_user_{i}_{unique_id}@company.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)
            users_to_create.append(user_id)
        
        assert len(users_to_create) == 3, "应创建3个用户"
        
        for user_id in users_to_create:
            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)
    
    async def test_bs_system_config_change(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-05: 系统配置变更流程
        
        业务场景:
        1. 修改系统配置
        2. 验证配置生效
        3. 配置回滚
        4. 验证回滚生效
        
        验收标准:
        - 配置修改成功
        - 新配置立即生效
        - 回滚操作成功
        - 回滚后配置正确
        """
        config_api = ConfigAPI(authenticated_client)
        
        unique_id = f"config_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        
        config_data = {
            "configKey": f"test_config_{unique_id}",
            "configName": f"测试配置_{unique_id}",
            "configType": "Y",
            "configValue": "initial_value",
            "remark": "业务场景测试-初始配置"
        }
        create_response = await config_api.create_config(config_data)
        assert create_response.status_code == 201
        config_id = create_response.json()["id"]
        test_data_manager.add_config(config_id)
        
        get_response = await config_api.get_config_by_key(f"test_config_{unique_id}")
        assert get_response.status_code == 200
        assert get_response.json()["configValue"] == "initial_value"
        
        update_data = {
            "configValue": "updated_value",
            "remark": "业务场景测试-更新配置"
        }
        update_response = await config_api.update_config(config_id, update_data)
        assert update_response.status_code == 200, "配置更新失败"
        
        verify_response = await config_api.get_config_by_key(f"test_config_{unique_id}")
        assert verify_response.json()["configValue"] == "updated_value", "配置更新未生效"
        
        rollback_data = {
            "configValue": "initial_value",
            "remark": "业务场景测试-配置回滚"
        }
        rollback_response = await config_api.update_config(config_id, rollback_data)
        assert rollback_response.status_code == 200, "配置回滚失败"
        
        final_response = await config_api.get_config_by_key(f"test_config_{unique_id}")
        assert final_response.json()["configValue"] == "initial_value", "配置回滚未生效"
        
        await config_api.delete_config(config_id)
        test_data_manager._configs.remove(config_id)
    
    async def test_bs_audit_trail_verification(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-06: 审计日志验证流程
        
        业务场景:
        1. 执行关键操作
        2. 查询审计日志
        3. 验证日志完整性
        4. 导出审计日志
        
        验收标准:
        - 操作被记录
        - 日志信息完整
        - 日志可查询
        - 日志可导出
        """
        user_api = UserAPI(authenticated_client)
        audit_api = AuditAPI(authenticated_client)
        
        unique_id = f"audit_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}"
        
        user_data = {
            "username": f"audit_user_{unique_id}",
            "password": "Test123!@#",
            "email": f"audit_{unique_id}@company.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)
        
        await asyncio.sleep(1)
        
        audit_response = await audit_api.get_audit_logs(
            page=0,
            size=10,
            module="用户管理",
            operation="创建用户"
        )
        assert audit_response.status_code == 200, "查询审计日志失败"
        
        audit_logs = audit_response.json()
        assert len(audit_logs) > 0, "应存在审计日志"
        
        latest_log = audit_logs[0]
        assert "operation" in latest_log, "日志应包含操作类型"
        assert "operator" in latest_log, "日志应包含操作人"
        assert "timestamp" in latest_log, "日志应包含时间戳"
        
        await user_api.delete_user(user_id)
        test_data_manager._users.remove(user_id)


@pytest.mark.uat
@pytest.mark.business_scenario
@pytest.mark.regression
class TestBusinessScenarioRegression:
    """业务场景回归测试"""
    
    @pytest.fixture
    async def authenticated_client(self):
        async with AuthAPI.create_client() as client:
            auth_api = AuthAPI(client)
            await auth_api.login(settings.TEST_USERNAME, settings.TEST_PASSWORD)
            yield client
    
    @pytest.fixture
    async def test_data_manager(self):
        from utils.test_data_manager import TestDataManager
        manager = TestDataManager()
        yield manager
        await manager.cleanup_all()
    
    @pytest.mark.asyncio
    async def test_bs_concurrent_user_operations(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-REG-01: 并发用户操作测试
        
        验证点:
        - 多用户并发创建
        - 数据一致性
        - 无死锁
        """
        user_api = UserAPI(authenticated_client)
        
        unique_id = f"concurrent_{int(time.time() * 1000)}"
        
        async def create_user(index: int):
            user_data = {
                "username": f"concurrent_user_{index}_{unique_id}",
                "password": "Test123!@#",
                "email": f"concurrent_{index}_{unique_id}@company.com",
                "status": 1
            }
            response = await user_api.create_user(user_data)
            return response
        
        tasks = [create_user(i) for i in range(5)]
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        success_count = sum(1 for r in results if not isinstance(r, Exception) and r.status_code == 201)
        assert success_count >= 3, f"至少应有3个用户创建成功，实际: {success_count}"
        
        for result in results:
            if not isinstance(result, Exception) and result.status_code == 201:
                user_id = result.json()["id"]
                await user_api.delete_user(user_id)
    
    @pytest.mark.asyncio
    async def test_bs_data_consistency(
        self, authenticated_client, test_data_manager
    ):
        """
        BS-REG-02: 数据一致性测试
        
        验证点:
        - 创建后立即查询
        - 数据一致性
        """
        user_api = UserAPI(authenticated_client)
        
        unique_id = f"consistency_{int(time.time() * 1000)}"
        
        user_data = {
            "username": f"consistency_user_{unique_id}",
            "password": "Test123!@#",
            "email": f"consistency_{unique_id}@company.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)
        
        get_response = await user_api.get_user_by_id(user_id)
        assert get_response.status_code == 200
        
        created_user = create_response.json()
        fetched_user = get_response.json()
        
        assert created_user["username"] == fetched_user["username"], "用户名应一致"
        assert created_user["email"] == fetched_user["email"], "邮箱应一致"
        
        await user_api.delete_user(user_id)
        test_data_manager._users.remove(user_id)