344 lines
12 KiB
Python
344 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
员工调动记录模块测试脚本
|
|
测试所有API接口功能
|
|
作者: ruoyi
|
|
日期: 2026-02-10
|
|
"""
|
|
|
|
import requests
|
|
import json
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, List, Any
|
|
|
|
# 配置
|
|
BASE_URL = "http://localhost:8080"
|
|
USERNAME = "admin"
|
|
PASSWORD = "admin123"
|
|
|
|
class TestStaffTransfer:
|
|
"""员工调动记录测试类"""
|
|
|
|
def __init__(self):
|
|
self.base_url = BASE_URL
|
|
self.token = None
|
|
self.headers = {}
|
|
self.test_results = []
|
|
self.created_ids = [] # 保存创建的记录ID,用于清理
|
|
|
|
def log_test(self, test_name: str, success: bool, message: str = ""):
|
|
"""记录测试结果"""
|
|
result = {
|
|
"test_name": test_name,
|
|
"success": success,
|
|
"message": message,
|
|
"timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
}
|
|
self.test_results.append(result)
|
|
status = "✓ PASS" if success else "✗ FAIL"
|
|
print(f"{status}: {test_name}")
|
|
if message:
|
|
print(f" -> {message}")
|
|
|
|
def login(self) -> bool:
|
|
"""登录获取token"""
|
|
try:
|
|
url = f"{self.base_url}/login/test"
|
|
data = {
|
|
"username": USERNAME,
|
|
"password": PASSWORD
|
|
}
|
|
response = requests.post(url, json=data)
|
|
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, result.get("msg", "登录失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("用户登录", False, str(e))
|
|
return False
|
|
|
|
def test_add_transfer(self, data: Dict[str, Any]) -> bool:
|
|
"""测试新增调动记录"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer"
|
|
response = requests.post(url, json=data, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
self.log_test("新增调动记录", True, f"成功创建记录")
|
|
return True
|
|
else:
|
|
self.log_test("新增调动记录", False, result.get("msg", "新增失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("新增调动记录", False, str(e))
|
|
return False
|
|
|
|
def test_list_transfer(self) -> bool:
|
|
"""测试查询调动记录列表"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/list"
|
|
params = {
|
|
"pageNum": 1,
|
|
"pageSize": 10
|
|
}
|
|
response = requests.get(url, params=params, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
rows = result.get("rows", [])
|
|
total = result.get("total", 0)
|
|
self.log_test("查询调动记录列表", True, f"查询到 {total} 条记录")
|
|
return True
|
|
else:
|
|
self.log_test("查询调动记录列表", False, result.get("msg", "查询失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("查询调动记录列表", False, str(e))
|
|
return False
|
|
|
|
def test_get_transfer(self, transfer_id: int) -> bool:
|
|
"""测试获取调动记录详情"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/{transfer_id}"
|
|
response = requests.get(url, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
data = result.get("data", {})
|
|
self.log_test("获取调动记录详情", True, f"获取记录详情成功")
|
|
return True
|
|
else:
|
|
self.log_test("获取调动记录详情", False, result.get("msg", "获取失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("获取调动记录详情", False, str(e))
|
|
return False
|
|
|
|
def test_update_transfer(self, data: Dict[str, Any]) -> bool:
|
|
"""测试修改调动记录"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer"
|
|
response = requests.put(url, json=data, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
self.log_test("修改调动记录", True, "修改记录成功")
|
|
return True
|
|
else:
|
|
self.log_test("修改调动记录", False, result.get("msg", "修改失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("修改调动记录", False, str(e))
|
|
return False
|
|
|
|
def test_delete_transfer(self, ids: List[int]) -> bool:
|
|
"""测试删除调动记录"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/{','.join(map(str, ids))}"
|
|
response = requests.delete(url, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
self.log_test("删除调动记录", True, f"删除记录成功")
|
|
return True
|
|
else:
|
|
self.log_test("删除调动记录", False, result.get("msg", "删除失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("删除调动记录", False, str(e))
|
|
return False
|
|
|
|
def test_export_transfer(self) -> bool:
|
|
"""测试导出调动记录"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/export"
|
|
response = requests.post(url, headers=self.headers)
|
|
|
|
if response.status_code == 200:
|
|
self.log_test("导出调动记录", True, f"导出成功,文件大小: {len(response.content)} bytes")
|
|
return True
|
|
else:
|
|
self.log_test("导出调动记录", False, f"导出失败,状态码: {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("导出调动记录", False, str(e))
|
|
return False
|
|
|
|
def test_import_template(self) -> bool:
|
|
"""测试下载导入模板"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/importTemplate"
|
|
response = requests.post(url, headers=self.headers)
|
|
|
|
if response.status_code == 200:
|
|
self.log_test("下载导入模板", True, f"下载成功,文件大小: {len(response.content)} bytes")
|
|
return True
|
|
else:
|
|
self.log_test("下载导入模板", False, f"下载失败,状态码: {response.status_code}")
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("下载导入模板", False, str(e))
|
|
return False
|
|
|
|
def test_import_data(self) -> bool:
|
|
"""测试导入数据(异步)"""
|
|
try:
|
|
# 这里需要准备一个测试Excel文件
|
|
# 由于无法直接上传文件,这里只测试接口是否可访问
|
|
url = f"{self.base_url}/ccdi/staffTransfer/importData"
|
|
|
|
# 注意: 实际测试时需要准备真实的Excel文件
|
|
self.log_test("导入数据(异步)", True, "接口需要Excel文件,跳过实际导入测试")
|
|
return True
|
|
except Exception as e:
|
|
self.log_test("导入数据(异步)", False, str(e))
|
|
return False
|
|
|
|
def test_import_status(self, task_id: str) -> bool:
|
|
"""测试查询导入状态"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/importStatus/{task_id}"
|
|
response = requests.get(url, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
self.log_test("查询导入状态", True, f"查询成功")
|
|
return True
|
|
else:
|
|
self.log_test("查询导入状态", False, result.get("msg", "查询失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("查询导入状态", False, str(e))
|
|
return False
|
|
|
|
def test_get_staff_list(self) -> bool:
|
|
"""测试获取员工列表"""
|
|
try:
|
|
url = f"{self.base_url}/ccdi/staffTransfer/staffList"
|
|
params = {"name": ""}
|
|
response = requests.get(url, params=params, headers=self.headers)
|
|
result = response.json()
|
|
|
|
if result.get("code") == 200:
|
|
data = result.get("data", [])
|
|
self.log_test("获取员工列表", True, f"获取到 {len(data)} 个员工")
|
|
return True
|
|
else:
|
|
self.log_test("获取员工列表", False, result.get("msg", "获取失败"))
|
|
return False
|
|
except Exception as e:
|
|
self.log_test("获取员工列表", False, str(e))
|
|
return False
|
|
|
|
def run_all_tests(self):
|
|
"""运行所有测试"""
|
|
print("=" * 60)
|
|
print("员工调动记录模块测试开始")
|
|
print("=" * 60)
|
|
|
|
# 1. 登录
|
|
if not self.login():
|
|
print("登录失败,终止测试")
|
|
return
|
|
|
|
# 2. 测试获取员工列表
|
|
self.test_get_staff_list()
|
|
|
|
# 3. 测试新增调动记录
|
|
add_data = {
|
|
"staffId": 1,
|
|
"transferType": "PROMOTION",
|
|
"transferSubType": "正常晋升",
|
|
"deptIdBefore": 100,
|
|
"deptNameBefore": "技术部",
|
|
"gradeBefore": "P5",
|
|
"positionBefore": "工程师",
|
|
"salaryLevelBefore": "L1",
|
|
"deptIdAfter": 101,
|
|
"deptNameAfter": "研发部",
|
|
"gradeAfter": "P6",
|
|
"positionAfter": "高级工程师",
|
|
"salaryLevelAfter": "L2",
|
|
"transferDate": "2026-02-10"
|
|
}
|
|
|
|
if self.test_add_transfer(add_data):
|
|
# 获取最新创建的记录ID
|
|
self.test_list_transfer()
|
|
|
|
# 4. 测试查询列表
|
|
self.test_list_transfer()
|
|
|
|
# 5. 测试获取详情 (假设ID为1)
|
|
self.test_get_transfer(1)
|
|
|
|
# 6. 测试修改
|
|
update_data = {
|
|
"id": 1,
|
|
"staffId": 1,
|
|
"transferType": "PROMOTION",
|
|
"transferSubType": "破格晋升",
|
|
"transferDate": "2026-02-10"
|
|
}
|
|
self.test_update_transfer(update_data)
|
|
|
|
# 7. 测试导出
|
|
self.test_export_transfer()
|
|
|
|
# 8. 测试下载导入模板
|
|
self.test_import_template()
|
|
|
|
# 9. 测试导入状态查询
|
|
self.test_import_status("test-task-id-123")
|
|
|
|
# 10. 生成测试报告
|
|
self.generate_report()
|
|
|
|
def generate_report(self):
|
|
"""生成测试报告"""
|
|
print("\n" + "=" * 60)
|
|
print("测试报告")
|
|
print("=" * 60)
|
|
|
|
total = len(self.test_results)
|
|
passed = sum(1 for r in self.test_results if r["success"])
|
|
failed = total - passed
|
|
|
|
print(f"总测试数: {total}")
|
|
print(f"通过: {passed}")
|
|
print(f"失败: {failed}")
|
|
print(f"通过率: {(passed/total*100):.2f}%")
|
|
|
|
# 保存测试报告到文件
|
|
report_path = "D:/ccdi/ccdi/doc/测试数据/员工调动记录/test_report.json"
|
|
try:
|
|
with open(report_path, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"summary": {
|
|
"total": total,
|
|
"passed": passed,
|
|
"failed": failed,
|
|
"pass_rate": f"{(passed/total*100):.2f}%"
|
|
},
|
|
"details": self.test_results
|
|
}, f, ensure_ascii=False, indent=2)
|
|
print(f"\n测试报告已保存到: {report_path}")
|
|
except Exception as e:
|
|
print(f"\n保存测试报告失败: {e}")
|
|
|
|
print("=" * 60)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
tester = TestStaffTransfer()
|
|
tester.run_all_tests()
|