#!/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()