ccdi/doc/测试数据/员工调动记录/test_unique_constraint.py

"""
员工调动记录唯一性约束测试脚本
测试功能：新增、编辑、导入时的唯一性校验
"""

import requests
import json
from datetime import datetime, timedelta
import time

# 配置
BASE_URL = "http://localhost:8080"
USERNAME = "admin"
PASSWORD = "admin123"

# 获取Token
def get_token():
    """获取登录token"""
    url = f"{BASE_URL}/login/test"
    data = {
        "username": USERNAME,
        "password": PASSWORD
    }
    response = requests.post(url, json=data)
    result = response.json()
    if result.get("code") == 200:
        return result.get("token")
    else:
        raise Exception(f"登录失败: {result}")

def get_headers(token):
    """获取请求头"""
    return {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

def print_test_case(name, description):
    """打印测试用例标题"""
    print(f"\n{'='*60}")
    print(f"测试用例: {name}")
    print(f"描述: {description}")
    print(f"{'='*60}")

def print_result(success, message, data=None):
    """打印测试结果"""
    status = "✓ PASS" if success else "✗ FAIL"
    print(f"\n结果: {status}")
    print(f"信息: {message}")
    if data:
        print(f"数据: {json.dumps(data, ensure_ascii=False, indent=2)}")

def test_add_normal_record(token):
    """测试1: 新增正常记录"""
    print_test_case("新增正常记录", "应该成功创建调动记录")

    url = f"{BASE_URL}/ccdi/staffTransfer"
    headers = get_headers(token)

    # 获取一个有效的员工ID和部门ID
    staff_id = 1  # 假设存在
    dept_before = 100
    dept_after = 101

    data = {
        "staffId": staff_id,
        "transferType": "平调",
        "transferSubType": "部门间调动",
        "deptIdBefore": dept_before,
        "deptNameBefore": "测试部门A",
        "gradeBefore": "职级A",
        "positionBefore": "岗位A",
        "salaryLevelBefore": "薪级A",
        "deptIdAfter": dept_after,
        "deptNameAfter": "测试部门B",
        "gradeAfter": "职级B",
        "positionAfter": "岗位B",
        "salaryLevelAfter": "薪级B",
        "transferDate": "2026-03-01"
    }

    response = requests.post(url, headers=headers, json=data)
    result = response.json()

    if result.get("code") == 200:
        print_result(True, "新增成功", result)
        return data  # 返回数据用于后续测试
    else:
        print_result(False, f"新增失败: {result.get('msg')}")
        return None

def test_add_duplicate_record(token, base_data):
    """测试2: 新增重复记录"""
    print_test_case("新增重复记录", "应该提示记录已存在")

    url = f"{BASE_URL}/ccdi/staffTransfer"
    headers = get_headers(token)

    # 使用与测试1相同的数据
    response = requests.post(url, headers=headers, json=base_data)
    result = response.json()

    if result.get("code") != 200 and "已存在" in result.get("msg", ""):
        print_result(True, "正确拦截重复记录", {"msg": result.get("msg")})
    else:
        print_result(False, f"未正确拦截重复记录: {result}")

def test_edit_non_key_fields(token):
    """测试3: 编辑非关键字段"""
    print_test_case("编辑非关键字段", "修改职级、岗位等非唯一键字段，应该成功")

    # 先查询一条记录
    list_url = f"{BASE_URL}/ccdi/staffTransfer/list"
    headers = get_headers(token)

    response = requests.get(list_url, headers=headers)
    result = response.json()

    if result.get("code") == 200 and result.get("rows"):
        record = result["rows"][0]
        record_id = record["id"]

        edit_url = f"{BASE_URL}/ccdi/staffTransfer"
        edit_data = {
            "id": record_id,
            "staffId": record["staffId"],
            "transferType": record["transferType"],
            "transferSubType": "修改后的子类型",
            "deptIdBefore": record["deptIdBefore"],
            "deptNameBefore": record["deptNameBefore"],
            "gradeBefore": "修改后的职级",
            "positionBefore": "修改后的岗位",
            "salaryLevelBefore": record["salaryLevelBefore"],
            "deptIdAfter": record["deptIdAfter"],
            "deptNameAfter": record["deptNameAfter"],
            "gradeAfter": "修改后的职级",
            "positionAfter": "修改后的岗位",
            "salaryLevelAfter": record["salaryLevelAfter"],
            "transferDate": record["transferDate"]
        }

        response = requests.put(edit_url, headers=headers, json=edit_data)
        result = response.json()

        if result.get("code") == 200:
            print_result(True, "编辑非关键字段成功")
        else:
            print_result(False, f"编辑失败: {result.get('msg')}")
    else:
        print_result(False, "没有可用的测试记录")

def test_edit_to_duplicate(token):
    """测试4: 编辑为重复记录"""
    print_test_case("编辑为重复记录", "修改唯一键字段导致重复，应该失败")

    # 需要先创建两条不同的记录，然后尝试将第一条编辑为与第二条重复
    # 这里简化处理：尝试修改日期为已存在的日期

    list_url = f"{BASE_URL}/ccdi/staffTransfer/list"
    headers = get_headers(token)

    response = requests.get(list_url, headers=headers)
    result = response.json()

    if result.get("code") == 200 and len(result.get("rows", [])) >= 2:
        record1 = result["rows"][0]
        record2 = result["rows"][1]

        edit_url = f"{BASE_URL}/ccdi/staffTransfer"
        edit_data = {
            "id": record1["id"],
            "staffId": record1["staffId"],
            "transferType": record1["transferType"],
            "deptIdBefore": record1["deptIdBefore"],
            "deptNameBefore": record1["deptNameBefore"],
            "deptIdAfter": record1["deptIdAfter"],
            "deptNameAfter": record1["deptNameAfter"],
            "transferDate": record2["transferDate"]  # 使用另一条记录的日期
        }

        response = requests.put(edit_url, headers=headers, json=edit_data)
        result = response.json()

        if result.get("code") != 200:
            print_result(True, "正确拦截编辑为重复", {"msg": result.get("msg")})
        else:
            print_result(False, "未正确拦截编辑为重复")
    else:
        print_result(False, "需要至少2条记录进行测试")

def run_all_tests():
    """运行所有测试"""
    print("\n" + "="*60)
    print("员工调动记录唯一性约束测试")
    print("="*60)

    try:
        # 获取token
        print("\n正在登录...")
        token = get_token()
        print("✓ 登录成功")

        # 测试1: 新增正常记录
        base_data = test_add_normal_record(token)
        time.sleep(1)

        # 测试2: 新增重复记录
        if base_data:
            test_add_duplicate_record(token, base_data)
        time.sleep(1)

        # 测试3: 编辑非关键字段
        test_edit_non_key_fields(token)
        time.sleep(1)

        # 测试4: 编辑为重复记录
        test_edit_to_duplicate(token)

        print("\n" + "="*60)
        print("所有测试完成！")
        print("="*60)

    except Exception as e:
        print(f"\n✗ 测试执行失败: {str(e)}")
        import traceback
        traceback.print_exc()

if __name__ == "__main__":
    run_all_tests()