424 lines
18 KiB
Python
424 lines
18 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
员工调动记录功能完整测试脚本
|
||
测试新的员工搜索接口和Treeselect部门选择功能
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
from datetime import datetime
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:8080"
|
||
USERNAME = "admin"
|
||
PASSWORD = "admin123"
|
||
|
||
class StaffTransferTester:
|
||
def __init__(self):
|
||
self.base_url = BASE_URL
|
||
self.token = None
|
||
self.headers = {}
|
||
self.test_results = []
|
||
|
||
def log_test(self, test_name, passed, message=""):
|
||
"""记录测试结果"""
|
||
status = "✅ PASS" if passed else "❌ FAIL"
|
||
result = {
|
||
"test": test_name,
|
||
"status": status,
|
||
"message": message,
|
||
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
self.test_results.append(result)
|
||
print(f"{status} - {test_name}")
|
||
if message:
|
||
print(f" {message}")
|
||
|
||
def login(self):
|
||
"""登录获取token"""
|
||
print("\n=== 测试1: 用户登录 ===")
|
||
try:
|
||
url = f"{self.base_url}/login/test"
|
||
data = {
|
||
"username": USERNAME,
|
||
"password": PASSWORD
|
||
}
|
||
response = requests.post(url, json=data)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
self.token = result.get("token")
|
||
self.headers = {"Authorization": f"Bearer {self.token}"}
|
||
self.log_test("用户登录", True, f"成功获取token: {self.token[:20]}...")
|
||
return True
|
||
else:
|
||
self.log_test("用户登录", False, f"登录失败: {result.get('msg')}")
|
||
return False
|
||
else:
|
||
self.log_test("用户登录", False, f"HTTP错误: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
self.log_test("用户登录", False, f"异常: {str(e)}")
|
||
return False
|
||
|
||
def test_staff_search_no_param(self):
|
||
"""测试员工搜索接口 - 不带参数"""
|
||
print("\n=== 测试2: 员工搜索（不带参数）===")
|
||
try:
|
||
url = f"{self.base_url}/ccdi/baseStaff/options"
|
||
response = requests.get(url, headers=self.headers)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
data = result.get("data", [])
|
||
self.log_test("员工搜索（不带参数）", True, f"返回{len(data)}条记录")
|
||
if data:
|
||
print(f" 示例数据: {json.dumps(data[0], ensure_ascii=False)}")
|
||
return data
|
||
else:
|
||
self.log_test("员工搜索（不带参数）", False, f"业务错误: {result.get('msg')}")
|
||
return []
|
||
else:
|
||
self.log_test("员工搜索（不带参数）", False, f"HTTP错误: {response.status_code}")
|
||
return []
|
||
except Exception as e:
|
||
self.log_test("员工搜索（不带参数）", False, f"异常: {str(e)}")
|
||
return []
|
||
|
||
def test_staff_search_by_id(self, staff_list):
|
||
"""测试员工搜索接口 - 按员工ID搜索"""
|
||
print("\n=== 测试3: 员工搜索（按员工ID）===")
|
||
if not staff_list:
|
||
self.log_test("员工搜索（按员工ID）", False, "无可用员工数据")
|
||
return None
|
||
|
||
try:
|
||
staff_id = staff_list[0]["staffId"]
|
||
url = f"{self.base_url}/ccdi/baseStaff/options"
|
||
params = {"query": str(staff_id)}
|
||
response = requests.get(url, headers=self.headers, params=params)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
data = result.get("data", [])
|
||
self.log_test("员工搜索（按员工ID）", True, f"搜索'{staff_id}'返回{len(data)}条记录")
|
||
return staff_list[0]
|
||
else:
|
||
self.log_test("员工搜索（按员工ID）", False, f"业务错误: {result.get('msg')}")
|
||
return None
|
||
else:
|
||
self.log_test("员工搜索（按员工ID）", False, f"HTTP错误: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
self.log_test("员工搜索（按员工ID）", False, f"异常: {str(e)}")
|
||
return None
|
||
|
||
def test_staff_search_by_name(self, staff_list):
|
||
"""测试员工搜索接口 - 按姓名搜索"""
|
||
print("\n=== 测试4: 员工搜索（按姓名）===")
|
||
if not staff_list:
|
||
self.log_test("员工搜索（按姓名）", False, "无可用员工数据")
|
||
return []
|
||
|
||
try:
|
||
# 获取第一个员工的姓名进行搜索
|
||
staff_name = staff_list[0]["staffName"]
|
||
url = f"{self.base_url}/ccdi/baseStaff/options"
|
||
params = {"query": staff_name}
|
||
response = requests.get(url, headers=self.headers, params=params)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
data = result.get("data", [])
|
||
self.log_test("员工搜索（按姓名）", True, f"搜索'{staff_name}'返回{len(data)}条记录")
|
||
if data:
|
||
print(f" 匹配结果: {json.dumps(data[0], ensure_ascii=False)}")
|
||
return data
|
||
else:
|
||
self.log_test("员工搜索（按姓名）", False, f"业务错误: {result.get('msg')}")
|
||
return []
|
||
else:
|
||
self.log_test("员工搜索（按姓名）", False, f"HTTP错误: {response.status_code}")
|
||
return []
|
||
except Exception as e:
|
||
self.log_test("员工搜索（按姓名）", False, f"异常: {str(e)}")
|
||
return []
|
||
|
||
def test_add_transfer(self, staff_data):
|
||
"""测试新增调动记录"""
|
||
print("\n=== 测试5: 新增员工调动记录 ===")
|
||
try:
|
||
url = f"{self.base_url}/ccdi/staffTransfer"
|
||
data = {
|
||
"staffId": staff_data["staffId"],
|
||
"transferType": "LATERAL",
|
||
"transferSubType": "测试调动",
|
||
"deptIdBefore": 100, # 示例部门ID
|
||
"deptNameBefore": "测试部门A",
|
||
"gradeBefore": "初级",
|
||
"positionBefore": "员工",
|
||
"salaryLevelBefore": "P1",
|
||
"deptIdAfter": 101, # 示例部门ID
|
||
"deptNameAfter": "测试部门B",
|
||
"gradeAfter": "中级",
|
||
"positionAfter": "主管",
|
||
"salaryLevelAfter": "P2",
|
||
"transferDate": datetime.now().strftime("%Y-%m-%d")
|
||
}
|
||
response = requests.post(url, json=data, headers=self.headers)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
self.log_test("新增员工调动记录", True, "新增成功")
|
||
return True
|
||
else:
|
||
self.log_test("新增员工调动记录", False, f"业务错误: {result.get('msg')}")
|
||
return False
|
||
else:
|
||
self.log_test("新增员工调动记录", False, f"HTTP错误: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
self.log_test("新增员工调动记录", False, f"异常: {str(e)}")
|
||
return False
|
||
|
||
def test_query_transfer_list(self):
|
||
"""测试查询调动记录列表"""
|
||
print("\n=== 测试6: 查询调动记录列表 ===")
|
||
try:
|
||
url = f"{self.base_url}/ccdi/staffTransfer/list"
|
||
params = {
|
||
"pageNum": 1,
|
||
"pageSize": 10
|
||
}
|
||
response = requests.get(url, headers=self.headers, params=params)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
rows = result.get("rows", [])
|
||
total = result.get("total", 0)
|
||
self.log_test("查询调动记录列表", True, f"查询成功，共{total}条记录，当前页{len(rows)}条")
|
||
if rows:
|
||
print(f" 示例记录ID: {rows[0].get('id', 'N/A')}")
|
||
return rows
|
||
else:
|
||
self.log_test("查询调动记录列表", False, f"业务错误: {result.get('msg')}")
|
||
return []
|
||
else:
|
||
self.log_test("查询调动记录列表", False, f"HTTP错误: {response.status_code}")
|
||
return []
|
||
except Exception as e:
|
||
self.log_test("查询调动记录列表", False, f"异常: {str(e)}")
|
||
return []
|
||
|
||
def test_get_transfer_detail(self, transfer_list):
|
||
"""测试获取调动记录详情"""
|
||
print("\n=== 测试7: 获取调动记录详情 ===")
|
||
if not transfer_list:
|
||
self.log_test("获取调动记录详情", False, "无可用调动记录")
|
||
return None
|
||
|
||
try:
|
||
transfer_id = transfer_list[0].get("id")
|
||
url = f"{self.base_url}/ccdi/staffTransfer/{transfer_id}"
|
||
response = requests.get(url, headers=self.headers)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
data = result.get("data")
|
||
self.log_test("获取调动记录详情", True, f"获取成功，记录ID: {transfer_id}")
|
||
print(f" 员工ID: {data.get('staffId')}")
|
||
print(f" 调动类型: {data.get('transferType')}")
|
||
return data
|
||
else:
|
||
self.log_test("获取调动记录详情", False, f"业务错误: {result.get('msg')}")
|
||
return None
|
||
else:
|
||
self.log_test("获取调动记录详情", False, f"HTTP错误: {response.status_code}")
|
||
return None
|
||
except Exception as e:
|
||
self.log_test("获取调动记录详情", False, f"异常: {str(e)}")
|
||
return None
|
||
|
||
def test_edit_transfer(self, transfer_list):
|
||
"""测试编辑调动记录"""
|
||
print("\n=== 测试8: 编辑员工调动记录 ===")
|
||
if not transfer_list:
|
||
self.log_test("编辑员工调动记录", False, "无可用调动记录")
|
||
return False
|
||
|
||
try:
|
||
transfer_id = transfer_list[0].get("id")
|
||
url = f"{self.base_url}/ccdi/staffTransfer"
|
||
data = {
|
||
"id": transfer_id,
|
||
"staffId": transfer_list[0].get("staffId"), # 员工ID不可修改，但需要传递
|
||
"transferType": "PROMOTION", # 修改调动类型
|
||
"transferSubType": "测试调动-已编辑",
|
||
"deptIdBefore": 100,
|
||
"deptNameBefore": "测试部门A",
|
||
"gradeBefore": "初级",
|
||
"positionBefore": "员工",
|
||
"salaryLevelBefore": "P1",
|
||
"deptIdAfter": 102,
|
||
"deptNameAfter": "测试部门C",
|
||
"gradeAfter": "高级",
|
||
"positionAfter": "经理",
|
||
"salaryLevelAfter": "P3",
|
||
"transferDate": datetime.now().strftime("%Y-%m-%d")
|
||
}
|
||
response = requests.put(url, json=data, headers=self.headers)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
self.log_test("编辑员工调动记录", True, "编辑成功")
|
||
return True
|
||
else:
|
||
self.log_test("编辑员工调动记录", False, f"业务错误: {result.get('msg')}")
|
||
return False
|
||
else:
|
||
self.log_test("编辑员工调动记录", False, f"HTTP错误: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
self.log_test("编辑员工调动记录", False, f"异常: {str(e)}")
|
||
return False
|
||
|
||
def test_delete_transfer(self, transfer_list):
|
||
"""测试删除调动记录"""
|
||
print("\n=== 测试9: 删除员工调动记录 ===")
|
||
if not transfer_list:
|
||
self.log_test("删除员工调动记录", False, "无可用调动记录")
|
||
return False
|
||
|
||
try:
|
||
transfer_id = transfer_list[0].get("id")
|
||
url = f"{self.base_url}/ccdi/staffTransfer/{transfer_id}"
|
||
response = requests.delete(url, headers=self.headers)
|
||
if response.status_code == 200:
|
||
result = response.json()
|
||
if result.get("code") == 200:
|
||
self.log_test("删除员工调动记录", True, f"删除成功，记录ID: {transfer_id}")
|
||
return True
|
||
else:
|
||
self.log_test("删除员工调动记录", False, f"业务错误: {result.get('msg')}")
|
||
return False
|
||
else:
|
||
self.log_test("删除员工调动记录", False, f"HTTP错误: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
self.log_test("删除员工调动记录", False, f"异常: {str(e)}")
|
||
return False
|
||
|
||
def test_export_transfer(self):
|
||
"""测试导出调动记录"""
|
||
print("\n=== 测试10: 导出调动记录 ===")
|
||
try:
|
||
url = f"{self.base_url}/ccdi/staffTransfer/export"
|
||
params = {
|
||
"staffId": "",
|
||
"transferType": "",
|
||
"deptIdAfter": ""
|
||
}
|
||
response = requests.post(url, headers=self.headers, params=params)
|
||
if response.status_code == 200:
|
||
# 检查是否返回Excel文件
|
||
content_type = response.headers.get("Content-Type", "")
|
||
if "excel" in content_type or "spreadsheet" in content_type or response.status_code == 200:
|
||
self.log_test("导出调动记录", True, f"导出成功，文件类型: {content_type}")
|
||
return True
|
||
else:
|
||
self.log_test("导出调动记录", False, f"返回格式错误: {content_type}")
|
||
return False
|
||
else:
|
||
self.log_test("导出调动记录", False, f"HTTP错误: {response.status_code}")
|
||
return False
|
||
except Exception as e:
|
||
self.log_test("导出调动记录", False, f"异常: {str(e)}")
|
||
return False
|
||
|
||
def generate_report(self):
|
||
"""生成测试报告"""
|
||
print("\n" + "="*60)
|
||
print("测试报告")
|
||
print("="*60)
|
||
|
||
passed = sum(1 for r in self.test_results if "PASS" in r["status"])
|
||
failed = sum(1 for r in self.test_results if "FAIL" in r["status"])
|
||
|
||
print(f"\n总计: {len(self.test_results)} 个测试")
|
||
print(f"通过: {passed} 个")
|
||
print(f"失败: {failed} 个")
|
||
if len(self.test_results) > 0:
|
||
print(f"成功率: {passed/len(self.test_results)*100:.1f}%\n")
|
||
|
||
# 保存到文件
|
||
report_path = "D:\\ccdi\\ccdi\\doc\\测试数据\\员工调动记录\\test_report.txt"
|
||
with open(report_path, "w", encoding="utf-8") as f:
|
||
f.write("员工调动记录功能测试报告\n")
|
||
f.write("="*60 + "\n")
|
||
f.write(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
|
||
f.write(f"总计: {len(self.test_results)} 个测试\n")
|
||
f.write(f"通过: {passed} 个\n")
|
||
f.write(f"失败: {failed} 个\n")
|
||
if len(self.test_results) > 0:
|
||
f.write(f"成功率: {passed/len(self.test_results)*100:.1f}%\n\n")
|
||
f.write("详细结果:\n")
|
||
f.write("-"*60 + "\n")
|
||
for result in self.test_results:
|
||
f.write(f"{result['status']} - {result['test']}\n")
|
||
if result['message']:
|
||
f.write(f" {result['message']}\n")
|
||
|
||
print(f"测试报告已保存至: {report_path}")
|
||
|
||
def run_all_tests(self):
|
||
"""运行所有测试"""
|
||
print("="*60)
|
||
print("员工调动记录功能测试")
|
||
print("="*60)
|
||
|
||
# 测试1: 登录
|
||
if not self.login():
|
||
print("登录失败，终止测试")
|
||
self.generate_report()
|
||
return
|
||
|
||
# 测试2: 搜索员工（不带参数）
|
||
staff_list = self.test_staff_search_no_param()
|
||
|
||
# 测试3: 搜索员工（按ID）
|
||
staff_data = self.test_staff_search_by_id(staff_list)
|
||
|
||
# 测试4: 搜索员工（按姓名）
|
||
self.test_staff_search_by_name(staff_list)
|
||
|
||
# 测试5: 新增调动记录
|
||
if staff_data:
|
||
add_success = self.test_add_transfer(staff_data)
|
||
else:
|
||
add_success = False
|
||
print(" 跳过新增测试：无有效员工数据")
|
||
|
||
# 测试6: 查询调动记录列表
|
||
transfer_list = self.test_query_transfer_list()
|
||
|
||
# 测试7: 获取调动记录详情
|
||
self.test_get_transfer_detail(transfer_list)
|
||
|
||
# 测试8: 编辑调动记录
|
||
if add_success and transfer_list:
|
||
self.test_edit_transfer(transfer_list)
|
||
|
||
# 测试9: 删除调动记录（可选，谨慎执行）
|
||
# self.test_delete_transfer(transfer_list)
|
||
|
||
# 测试10: 导出调动记录
|
||
self.test_export_transfer()
|
||
|
||
# 生成报告
|
||
self.generate_report()
|
||
|
||
if __name__ == "__main__":
|
||
tester = StaffTransferTester()
|
||
tester.run_all_tests()
|