929 lines
33 KiB
Python
929 lines
33 KiB
Python
|
#!/usr/bin/env python
|
||
|
|
# -*- coding: utf-8 -*-
|
||
|
|
"""
|
||
|
|
导入文件内部主键重复检测功能测试脚本
|
||
|
|
|
||
|
|
测试目标:
|
||
|
|
1. 采购交易导入 - Excel内采购事项ID重复检测
|
||
|
|
2. 员工信息导入 - Excel内柜员号和身份证号重复检测
|
||
|
|
|
||
|
|
作者: 测试专家
|
||
|
|
日期: 2026-02-09
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
import json
|
||
|
|
import requests
|
||
|
|
from openpyxl import Workbook
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from typing import List, Dict, Tuple
|
||
|
|
|
||
|
|
# ==================== 配置部分 ====================
|
||
|
|
BASE_URL = "http://localhost:8080"
|
||
|
|
LOGIN_URL = f"{BASE_URL}/login/test"
|
||
|
|
|
||
|
|
# 测试账号
|
||
|
|
USERNAME = "admin"
|
||
|
|
PASSWORD = "admin123"
|
||
|
|
|
||
|
|
# 测试结果保存目录
|
||
|
|
REPORT_DIR = "D:/ccdi/ccdi/doc/test-reports"
|
||
|
|
EXCEL_DIR = "D:/ccdi/ccdi/doc/test-data/temp"
|
||
|
|
|
||
|
|
# 创建必要目录
|
||
|
|
os.makedirs(REPORT_DIR, exist_ok=True)
|
||
|
|
os.makedirs(EXCEL_DIR, exist_ok=True)
|
||
|
|
|
||
|
|
|
||
|
|
class APIClient:
|
||
|
|
"""API客户端"""
|
||
|
|
|
||
|
|
def __init__(self, base_url: str):
|
||
|
|
self.base_url = base_url
|
||
|
|
self.token = None
|
||
|
|
self.session = requests.Session()
|
||
|
|
|
||
|
|
def login(self, username: str, password: str) -> bool:
|
||
|
|
"""登录获取token"""
|
||
|
|
try:
|
||
|
|
response = self.session.post(
|
||
|
|
LOGIN_URL,
|
||
|
|
json={"username": username, "password": password},
|
||
|
|
timeout=10
|
||
|
|
)
|
||
|
|
result = response.json()
|
||
|
|
|
||
|
|
if result.get("code") == 200:
|
||
|
|
self.token = result.get("data")
|
||
|
|
print(f"✓ 登录成功, Token: {self.token[:20]}...")
|
||
|
|
return True
|
||
|
|
else:
|
||
|
|
print(f"✗ 登录失败: {result.get('msg')}")
|
||
|
|
return False
|
||
|
|
except Exception as e:
|
||
|
|
print(f"✗ 登录异常: {str(e)}")
|
||
|
|
return False
|
||
|
|
|
||
|
|
def get_headers(self) -> Dict:
|
||
|
|
"""获取请求头"""
|
||
|
|
return {
|
||
|
|
"Authorization": f"Bearer {self.token}",
|
||
|
|
"Content-Type": "application/json"
|
||
|
|
}
|
||
|
|
|
||
|
|
def upload_file(self, url: str, file_path: str) -> Dict:
|
||
|
|
"""上传文件"""
|
||
|
|
try:
|
||
|
|
with open(file_path, 'rb') as f:
|
||
|
|
files = {'file': (os.path.basename(file_path), f, 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet')}
|
||
|
|
headers = {"Authorization": f"Bearer {self.token}"}
|
||
|
|
|
||
|
|
response = self.session.post(url, files=files, headers=headers, timeout=30)
|
||
|
|
return response.json()
|
||
|
|
except Exception as e:
|
||
|
|
return {"code": 500, "msg": f"上传失败: {str(e)}"}
|
||
|
|
|
||
|
|
def get_import_status(self, url: str) -> Dict:
|
||
|
|
"""查询导入状态"""
|
||
|
|
try:
|
||
|
|
response = self.session.get(url, headers=self.get_headers(), timeout=10)
|
||
|
|
return response.json()
|
||
|
|
except Exception as e:
|
||
|
|
return {"code": 500, "msg": f"查询状态失败: {str(e)}"}
|
||
|
|
|
||
|
|
def get_import_failures(self, url: str) -> Dict:
|
||
|
|
"""查询导入失败记录"""
|
||
|
|
try:
|
||
|
|
response = self.session.get(url, headers=self.get_headers(), timeout=10)
|
||
|
|
return response.json()
|
||
|
|
except Exception as e:
|
||
|
|
return {"code": 500, "msg": f"查询失败记录失败: {str(e)}"}
|
||
|
|
|
||
|
|
|
||
|
|
class ExcelGenerator:
|
||
|
|
"""Excel测试数据生成器"""
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def create_purchase_duplicate_data() -> str:
|
||
|
|
"""
|
||
|
|
场景1: Excel内采购事项ID重复
|
||
|
|
3条记录,采购事项ID都是 PURCHASE001
|
||
|
|
"""
|
||
|
|
wb = Workbook()
|
||
|
|
ws = wb.active
|
||
|
|
ws.title = "采购交易重复测试"
|
||
|
|
|
||
|
|
# 表头
|
||
|
|
headers = [
|
||
|
|
"采购事项ID", "采购类别", "项目名称", "标的物名称", "标的物描述",
|
||
|
|
"采购数量", "预算金额", "采购方式", "采购申请日期",
|
||
|
|
"申请人工号", "申请人姓名", "申请部门"
|
||
|
|
]
|
||
|
|
ws.append(headers)
|
||
|
|
|
||
|
|
# 测试数据 - 3条相同采购事项ID
|
||
|
|
base_date = datetime.now()
|
||
|
|
|
||
|
|
for i in range(3):
|
||
|
|
apply_date = base_date + timedelta(days=i)
|
||
|
|
ws.append([
|
||
|
|
f"PURCHASE001", # 相同的采购事项ID
|
||
|
|
f"采购类别{i+1}",
|
||
|
|
f"项目名称{i+1}",
|
||
|
|
f"标的物名称{i+1}",
|
||
|
|
f"标的物描述{i+1}",
|
||
|
|
10 + i,
|
||
|
|
10000.00 + i * 1000,
|
||
|
|
"公开招标",
|
||
|
|
apply_date.strftime("%Y-%m-%d"),
|
||
|
|
"1000001",
|
||
|
|
"张三",
|
||
|
|
"技术部"
|
||
|
|
])
|
||
|
|
|
||
|
|
file_path = os.path.join(EXCEL_DIR, "purchase_duplicate.xlsx")
|
||
|
|
wb.save(file_path)
|
||
|
|
return file_path
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def create_employee_employee_id_duplicate() -> str:
|
||
|
|
"""
|
||
|
|
场景2: Excel内员工柜员号重复
|
||
|
|
3条记录,柜员号都是 10001,身份证号不同
|
||
|
|
"""
|
||
|
|
wb = Workbook()
|
||
|
|
ws = wb.active
|
||
|
|
ws.title = "员工柜员号重复测试"
|
||
|
|
|
||
|
|
# 表头
|
||
|
|
headers = ["姓名", "柜员号", "所属部门ID", "身份证号", "电话", "入职时间", "状态"]
|
||
|
|
ws.append(headers)
|
||
|
|
|
||
|
|
# 测试数据 - 3条相同柜员号
|
||
|
|
for i in range(3):
|
||
|
|
ws.append([
|
||
|
|
f"员工{i+1}",
|
||
|
|
10001, # 相同的柜员号
|
||
|
|
103,
|
||
|
|
f"110101199001011{234+i}", # 不同的身份证号
|
||
|
|
f"1380000000{i}",
|
||
|
|
"2024-01-01",
|
||
|
|
"0"
|
||
|
|
])
|
||
|
|
|
||
|
|
file_path = os.path.join(EXCEL_DIR, "employee_employee_id_duplicate.xlsx")
|
||
|
|
wb.save(file_path)
|
||
|
|
return file_path
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def create_employee_id_card_duplicate() -> str:
|
||
|
|
"""
|
||
|
|
场景3: Excel内员工身份证号重复
|
||
|
|
3条记录,柜员号不同,身份证号相同
|
||
|
|
"""
|
||
|
|
wb = Workbook()
|
||
|
|
ws = wb.active
|
||
|
|
ws.title = "员工身份证号重复测试"
|
||
|
|
|
||
|
|
# 表头
|
||
|
|
headers = ["姓名", "柜员号", "所属部门ID", "身份证号", "电话", "入职时间", "状态"]
|
||
|
|
ws.append(headers)
|
||
|
|
|
||
|
|
# 测试数据 - 3条相同身份证号
|
||
|
|
for i in range(3):
|
||
|
|
ws.append([
|
||
|
|
f"员工{i+1}",
|
||
|
|
10001 + i, # 不同的柜员号
|
||
|
|
103,
|
||
|
|
"110101199001011234", # 相同的身份证号
|
||
|
|
f"1380000000{i}",
|
||
|
|
"2024-01-01",
|
||
|
|
"0"
|
||
|
|
])
|
||
|
|
|
||
|
|
file_path = os.path.join(EXCEL_DIR, "employee_id_card_duplicate.xlsx")
|
||
|
|
wb.save(file_path)
|
||
|
|
return file_path
|
||
|
|
|
||
|
|
@staticmethod
|
||
|
|
def create_mixed_duplicate_scenario() -> Tuple[str, str]:
|
||
|
|
"""
|
||
|
|
场景4: 混合重复(数据库+Excel)
|
||
|
|
- 第1条: 数据库中已存在
|
||
|
|
- 第2条: 全新数据
|
||
|
|
- 第3条: 与第2条Excel内重复
|
||
|
|
- 第4条: 全新数据
|
||
|
|
"""
|
||
|
|
# 采购交易混合重复数据
|
||
|
|
wb_purchase = Workbook()
|
||
|
|
ws_purchase = wb_purchase.active
|
||
|
|
ws_purchase.title = "采购混合重复测试"
|
||
|
|
|
||
|
|
headers = [
|
||
|
|
"采购事项ID", "采购类别", "项目名称", "标的物名称", "标的物描述",
|
||
|
|
"采购数量", "预算金额", "采购方式", "采购申请日期",
|
||
|
|
"申请人工号", "申请人姓名", "申请部门"
|
||
|
|
]
|
||
|
|
ws_purchase.append(headers)
|
||
|
|
|
||
|
|
base_date = datetime.now()
|
||
|
|
|
||
|
|
# 第1条: 数据库中已存在(需要先手动插入数据库)
|
||
|
|
ws_purchase.append([
|
||
|
|
"EXIST001", # 假设数据库中已存在
|
||
|
|
"采购类别1",
|
||
|
|
"项目名称1",
|
||
|
|
"标的物名称1",
|
||
|
|
"标的物描述1",
|
||
|
|
10,
|
||
|
|
10000.00,
|
||
|
|
"公开招标",
|
||
|
|
base_date.strftime("%Y-%m-%d"),
|
||
|
|
"1000001",
|
||
|
|
"张三",
|
||
|
|
"技术部"
|
||
|
|
])
|
||
|
|
|
||
|
|
# 第2条: 全新数据
|
||
|
|
ws_purchase.append([
|
||
|
|
"NEW001", # 新的采购事项ID
|
||
|
|
"采购类别2",
|
||
|
|
"项目名称2",
|
||
|
|
"标的物名称2",
|
||
|
|
"标的物描述2",
|
||
|
|
20,
|
||
|
|
20000.00,
|
||
|
|
"邀请招标",
|
||
|
|
(base_date + timedelta(days=1)).strftime("%Y-%m-%d"),
|
||
|
|
"1000002",
|
||
|
|
"李四",
|
||
|
|
"市场部"
|
||
|
|
])
|
||
|
|
|
||
|
|
# 第3条: 与第2条Excel内重复
|
||
|
|
ws_purchase.append([
|
||
|
|
"NEW001", # 与第2条重复
|
||
|
|
"采购类别3",
|
||
|
|
"项目名称3",
|
||
|
|
"标的物名称3",
|
||
|
|
"标的物描述3",
|
||
|
|
30,
|
||
|
|
30000.00,
|
||
|
|
"竞争性谈判",
|
||
|
|
(base_date + timedelta(days=2)).strftime("%Y-%m-%d"),
|
||
|
|
"1000003",
|
||
|
|
"王五",
|
||
|
|
"财务部"
|
||
|
|
])
|
||
|
|
|
||
|
|
# 第4条: 全新数据
|
||
|
|
ws_purchase.append([
|
||
|
|
"NEW002", # 新的采购事项ID
|
||
|
|
"采购类别4",
|
||
|
|
"项目名称4",
|
||
|
|
"标的物名称4",
|
||
|
|
"标的物描述4",
|
||
|
|
40,
|
||
|
|
40000.00,
|
||
|
|
"单一来源",
|
||
|
|
(base_date + timedelta(days=3)).strftime("%Y-%m-%d"),
|
||
|
|
"1000004",
|
||
|
|
"赵六",
|
||
|
|
"人事部"
|
||
|
|
])
|
||
|
|
|
||
|
|
purchase_file = os.path.join(EXCEL_DIR, "purchase_mixed_duplicate.xlsx")
|
||
|
|
wb_purchase.save(purchase_file)
|
||
|
|
|
||
|
|
# 员工混合重复数据
|
||
|
|
wb_employee = Workbook()
|
||
|
|
ws_employee = wb_employee.active
|
||
|
|
ws_employee.title = "员工混合重复测试"
|
||
|
|
|
||
|
|
headers = ["姓名", "柜员号", "所属部门ID", "身份证号", "电话", "入职时间", "状态"]
|
||
|
|
ws_employee.append(headers)
|
||
|
|
|
||
|
|
# 第1条: 数据库中已存在(假设柜员号99999已存在)
|
||
|
|
ws_employee.append([
|
||
|
|
"已存在员工",
|
||
|
|
99999, # 假设数据库中已存在
|
||
|
|
103,
|
||
|
|
"110101199001019999",
|
||
|
|
"13900000000",
|
||
|
|
"2024-01-01",
|
||
|
|
"0"
|
||
|
|
])
|
||
|
|
|
||
|
|
# 第2条: 全新数据
|
||
|
|
ws_employee.append([
|
||
|
|
"新员工1",
|
||
|
|
90001, # 新柜员号
|
||
|
|
103,
|
||
|
|
"110101199001011111",
|
||
|
|
"13800000001",
|
||
|
|
"2024-01-01",
|
||
|
|
"0"
|
||
|
|
])
|
||
|
|
|
||
|
|
# 第3条: 与第2条Excel内重复(柜员号重复)
|
||
|
|
ws_employee.append([
|
||
|
|
"新员工2",
|
||
|
|
90001, # 与第2条柜员号重复
|
||
|
|
103,
|
||
|
|
"110101199001012222",
|
||
|
|
"13800000002",
|
||
|
|
"2024-01-01",
|
||
|
|
"0"
|
||
|
|
])
|
||
|
|
|
||
|
|
# 第4条: 全新数据
|
||
|
|
ws_employee.append([
|
||
|
|
"新员工3",
|
||
|
|
90002, # 新柜员号
|
||
|
|
103,
|
||
|
|
"110101199001013333",
|
||
|
|
"13800000003",
|
||
|
|
"2024-01-01",
|
||
|
|
"0"
|
||
|
|
])
|
||
|
|
|
||
|
|
employee_file = os.path.join(EXCEL_DIR, "employee_mixed_duplicate.xlsx")
|
||
|
|
wb_employee.save(employee_file)
|
||
|
|
|
||
|
|
return purchase_file, employee_file
|
||
|
|
|
||
|
|
|
||
|
|
class TestCase:
|
||
|
|
"""测试用例基类"""
|
||
|
|
|
||
|
|
def __init__(self, name: str, description: str):
|
||
|
|
self.name = name
|
||
|
|
self.description = description
|
||
|
|
self.start_time = None
|
||
|
|
self.end_time = None
|
||
|
|
self.passed = False
|
||
|
|
self.error_message = None
|
||
|
|
self.details = {}
|
||
|
|
|
||
|
|
def run(self, client: APIClient):
|
||
|
|
"""运行测试用例"""
|
||
|
|
raise NotImplementedError
|
||
|
|
|
||
|
|
def to_dict(self) -> Dict:
|
||
|
|
"""转换为字典"""
|
||
|
|
return {
|
||
|
|
"name": self.name,
|
||
|
|
"description": self.description,
|
||
|
|
"passed": self.passed,
|
||
|
|
"error_message": self.error_message,
|
||
|
|
"details": self.details,
|
||
|
|
"duration": f"{(self.end_time - self.start_time).total_seconds():.2f}s" if self.start_time and self.end_time else "N/A"
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class PurchaseDuplicateTestCase(TestCase):
|
||
|
|
"""场景1: Excel内采购事项ID重复测试"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__(
|
||
|
|
"采购交易 - Excel内采购事项ID重复",
|
||
|
|
"测试导入3条采购事项ID相同的记录,预期第1条成功,第2、3条失败"
|
||
|
|
)
|
||
|
|
|
||
|
|
def run(self, client: APIClient):
|
||
|
|
self.start_time = datetime.now()
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 生成测试数据
|
||
|
|
file_path = ExcelGenerator.create_purchase_duplicate_data()
|
||
|
|
print(f" ✓ 生成测试数据: {file_path}")
|
||
|
|
|
||
|
|
# 上传文件
|
||
|
|
upload_url = f"{BASE_URL}/ccdi/purchaseTransaction/importData"
|
||
|
|
upload_result = client.upload_file(upload_url, file_path)
|
||
|
|
|
||
|
|
if upload_result.get("code") != 200:
|
||
|
|
self.error_message = f"上传失败: {upload_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
task_id = upload_result.get("data", {}).get("taskId")
|
||
|
|
print(f" ✓ 上传成功, TaskID: {task_id}")
|
||
|
|
|
||
|
|
# 等待异步任务完成
|
||
|
|
time.sleep(3)
|
||
|
|
|
||
|
|
# 查询导入状态
|
||
|
|
status_url = f"{BASE_URL}/ccdi/purchaseTransaction/importStatus/{task_id}"
|
||
|
|
status_result = client.get_import_status(status_url)
|
||
|
|
|
||
|
|
if status_result.get("code") != 200:
|
||
|
|
self.error_message = f"查询状态失败: {status_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
status_data = status_result.get("data", {})
|
||
|
|
print(f" ✓ 导入状态: {status_data}")
|
||
|
|
|
||
|
|
# 查询失败记录
|
||
|
|
failures_url = f"{BASE_URL}/ccdi/purchaseTransaction/importFailures/{task_id}"
|
||
|
|
failures_result = client.get_import_failures(failures_url)
|
||
|
|
|
||
|
|
if failures_result.get("code") != 200:
|
||
|
|
self.error_message = f"查询失败记录失败: {failures_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
failures = failures_result.get("rows", [])
|
||
|
|
|
||
|
|
# 验证结果
|
||
|
|
# 预期: 成功1条,失败2条
|
||
|
|
expected_success = 1
|
||
|
|
expected_failure = 2
|
||
|
|
actual_success = status_data.get("successCount", 0)
|
||
|
|
actual_failure = status_data.get("failureCount", 0)
|
||
|
|
|
||
|
|
self.details = {
|
||
|
|
"expected_success": expected_success,
|
||
|
|
"expected_failure": expected_failure,
|
||
|
|
"actual_success": actual_success,
|
||
|
|
"actual_failure": actual_failure,
|
||
|
|
"failures": failures
|
||
|
|
}
|
||
|
|
|
||
|
|
# 验证成功/失败数量
|
||
|
|
if actual_success != expected_success or actual_failure != expected_failure:
|
||
|
|
self.error_message = f"数量不匹配: 预期成功{expected_success}失败{expected_failure}, 实际成功{actual_success}失败{actual_failure}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
# 验证失败消息
|
||
|
|
if len(failures) < 2:
|
||
|
|
self.error_message = f"失败记录数量不足: 预期2条, 实际{len(failures)}条"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
# 检查失败消息是否包含"在导入文件中重复"
|
||
|
|
error_msg_1 = failures[0].get("errorMessage", "")
|
||
|
|
error_msg_2 = failures[1].get("errorMessage", "")
|
||
|
|
|
||
|
|
if "在导入文件中重复" not in error_msg_1 or "在导入文件中重复" not in error_msg_2:
|
||
|
|
self.error_message = f"错误消息不正确: {error_msg_1}, {error_msg_2}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
self.passed = True
|
||
|
|
print(f" ✓ 测试通过")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
self.error_message = f"测试异常: {str(e)}"
|
||
|
|
print(f" ✗ 测试异常: {str(e)}")
|
||
|
|
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
|
||
|
|
|
||
|
|
class EmployeeEmployeeIdDuplicateTestCase(TestCase):
|
||
|
|
"""场景2: Excel内员工柜员号重复测试"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__(
|
||
|
|
"员工信息 - Excel内柜员号重复",
|
||
|
|
"测试导入3条柜员号相同的记录,预期第1条成功,第2、3条失败"
|
||
|
|
)
|
||
|
|
|
||
|
|
def run(self, client: APIClient):
|
||
|
|
self.start_time = datetime.now()
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 生成测试数据
|
||
|
|
file_path = ExcelGenerator.create_employee_employee_id_duplicate()
|
||
|
|
print(f" ✓ 生成测试数据: {file_path}")
|
||
|
|
|
||
|
|
# 上传文件
|
||
|
|
upload_url = f"{BASE_URL}/ccdi/employee/importData"
|
||
|
|
upload_result = client.upload_file(upload_url, file_path)
|
||
|
|
|
||
|
|
if upload_result.get("code") != 200:
|
||
|
|
self.error_message = f"上传失败: {upload_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
task_id = upload_result.get("data", {}).get("taskId")
|
||
|
|
print(f" ✓ 上传成功, TaskID: {task_id}")
|
||
|
|
|
||
|
|
# 等待异步任务完成
|
||
|
|
time.sleep(3)
|
||
|
|
|
||
|
|
# 查询导入状态
|
||
|
|
status_url = f"{BASE_URL}/ccdi/employee/importStatus/{task_id}"
|
||
|
|
status_result = client.get_import_status(status_url)
|
||
|
|
|
||
|
|
if status_result.get("code") != 200:
|
||
|
|
self.error_message = f"查询状态失败: {status_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
status_data = status_result.get("data", {})
|
||
|
|
print(f" ✓ 导入状态: {status_data}")
|
||
|
|
|
||
|
|
# 查询失败记录
|
||
|
|
failures_url = f"{BASE_URL}/ccdi/employee/importFailures/{task_id}"
|
||
|
|
failures_result = client.get_import_failures(failures_url)
|
||
|
|
|
||
|
|
if failures_result.get("code") != 200:
|
||
|
|
self.error_message = f"查询失败记录失败: {failures_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
failures = failures_result.get("rows", [])
|
||
|
|
|
||
|
|
# 验证结果
|
||
|
|
expected_success = 1
|
||
|
|
expected_failure = 2
|
||
|
|
actual_success = status_data.get("successCount", 0)
|
||
|
|
actual_failure = status_data.get("failureCount", 0)
|
||
|
|
|
||
|
|
self.details = {
|
||
|
|
"expected_success": expected_success,
|
||
|
|
"expected_failure": expected_failure,
|
||
|
|
"actual_success": actual_success,
|
||
|
|
"actual_failure": actual_failure,
|
||
|
|
"failures": failures
|
||
|
|
}
|
||
|
|
|
||
|
|
if actual_success != expected_success or actual_failure != expected_failure:
|
||
|
|
self.error_message = f"数量不匹配: 预期成功{expected_success}失败{expected_failure}, 实际成功{actual_success}失败{actual_failure}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
if len(failures) < 2:
|
||
|
|
self.error_message = f"失败记录数量不足: 预期2条, 实际{len(failures)}条"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
# 验证失败消息
|
||
|
|
error_msg_1 = failures[0].get("errorMessage", "")
|
||
|
|
error_msg_2 = failures[1].get("errorMessage", "")
|
||
|
|
|
||
|
|
if "柜员号" not in error_msg_1 or "在导入文件中重复" not in error_msg_1:
|
||
|
|
self.error_message = f"错误消息不正确(第1条): {error_msg_1}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
if "柜员号" not in error_msg_2 or "在导入文件中重复" not in error_msg_2:
|
||
|
|
self.error_message = f"错误消息不正确(第2条): {error_msg_2}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
self.passed = True
|
||
|
|
print(f" ✓ 测试通过")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
self.error_message = f"测试异常: {str(e)}"
|
||
|
|
print(f" ✗ 测试异常: {str(e)}")
|
||
|
|
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
|
||
|
|
|
||
|
|
class EmployeeIdCardDuplicateTestCase(TestCase):
|
||
|
|
"""场景3: Excel内员工身份证号重复测试"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__(
|
||
|
|
"员工信息 - Excel内身份证号重复",
|
||
|
|
"测试导入3条身份证号相同的记录,预期第1条成功,第2、3条失败"
|
||
|
|
)
|
||
|
|
|
||
|
|
def run(self, client: APIClient):
|
||
|
|
self.start_time = datetime.now()
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 生成测试数据
|
||
|
|
file_path = ExcelGenerator.create_employee_id_card_duplicate()
|
||
|
|
print(f" ✓ 生成测试数据: {file_path}")
|
||
|
|
|
||
|
|
# 上传文件
|
||
|
|
upload_url = f"{BASE_URL}/ccdi/employee/importData"
|
||
|
|
upload_result = client.upload_file(upload_url, file_path)
|
||
|
|
|
||
|
|
if upload_result.get("code") != 200:
|
||
|
|
self.error_message = f"上传失败: {upload_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
task_id = upload_result.get("data", {}).get("taskId")
|
||
|
|
print(f" ✓ 上传成功, TaskID: {task_id}")
|
||
|
|
|
||
|
|
# 等待异步任务完成
|
||
|
|
time.sleep(3)
|
||
|
|
|
||
|
|
# 查询导入状态
|
||
|
|
status_url = f"{BASE_URL}/ccdi/employee/importStatus/{task_id}"
|
||
|
|
status_result = client.get_import_status(status_url)
|
||
|
|
|
||
|
|
if status_result.get("code") != 200:
|
||
|
|
self.error_message = f"查询状态失败: {status_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
status_data = status_result.get("data", {})
|
||
|
|
print(f" ✓ 导入状态: {status_data}")
|
||
|
|
|
||
|
|
# 查询失败记录
|
||
|
|
failures_url = f"{BASE_URL}/ccdi/employee/importFailures/{task_id}"
|
||
|
|
failures_result = client.get_import_failures(failures_url)
|
||
|
|
|
||
|
|
if failures_result.get("code") != 200:
|
||
|
|
self.error_message = f"查询失败记录失败: {failures_result.get('msg')}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
failures = failures_result.get("rows", [])
|
||
|
|
|
||
|
|
# 验证结果
|
||
|
|
expected_success = 1
|
||
|
|
expected_failure = 2
|
||
|
|
actual_success = status_data.get("successCount", 0)
|
||
|
|
actual_failure = status_data.get("failureCount", 0)
|
||
|
|
|
||
|
|
self.details = {
|
||
|
|
"expected_success": expected_success,
|
||
|
|
"expected_failure": expected_failure,
|
||
|
|
"actual_success": actual_success,
|
||
|
|
"actual_failure": actual_failure,
|
||
|
|
"failures": failures
|
||
|
|
}
|
||
|
|
|
||
|
|
if actual_success != expected_success or actual_failure != expected_failure:
|
||
|
|
self.error_message = f"数量不匹配: 预期成功{expected_success}失败{expected_failure}, 实际成功{actual_success}失败{actual_failure}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
if len(failures) < 2:
|
||
|
|
self.error_message = f"失败记录数量不足: 预期2条, 实际{len(failures)}条"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
# 验证失败消息
|
||
|
|
error_msg_1 = failures[0].get("errorMessage", "")
|
||
|
|
error_msg_2 = failures[1].get("errorMessage", "")
|
||
|
|
|
||
|
|
if "身份证号" not in error_msg_1 or "在导入文件中重复" not in error_msg_1:
|
||
|
|
self.error_message = f"错误消息不正确(第1条): {error_msg_1}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
if "身份证号" not in error_msg_2 or "在导入文件中重复" not in error_msg_2:
|
||
|
|
self.error_message = f"错误消息不正确(第2条): {error_msg_2}"
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
return
|
||
|
|
|
||
|
|
self.passed = True
|
||
|
|
print(f" ✓ 测试通过")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
self.error_message = f"测试异常: {str(e)}"
|
||
|
|
print(f" ✗ 测试异常: {str(e)}")
|
||
|
|
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
|
||
|
|
|
||
|
|
class MixedDuplicateTestCase(TestCase):
|
||
|
|
"""场景4: 混合重复(数据库+Excel)测试"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
super().__init__(
|
||
|
|
"混合重复 - 数据库+Excel重复",
|
||
|
|
"测试数据库已存在+Excel内重复的混合场景"
|
||
|
|
)
|
||
|
|
|
||
|
|
def run(self, client: APIClient):
|
||
|
|
self.start_time = datetime.now()
|
||
|
|
|
||
|
|
try:
|
||
|
|
# 生成测试数据
|
||
|
|
purchase_file, employee_file = ExcelGenerator.create_mixed_duplicate_scenario()
|
||
|
|
print(f" ✓ 生成测试数据: {purchase_file}, {employee_file}")
|
||
|
|
|
||
|
|
# 测试采购交易
|
||
|
|
print("\n >> 测试采购交易混合重复")
|
||
|
|
purchase_upload_url = f"{BASE_URL}/ccdi/purchaseTransaction/importData"
|
||
|
|
purchase_upload_result = client.upload_file(purchase_upload_url, purchase_file)
|
||
|
|
|
||
|
|
purchase_passed = False
|
||
|
|
purchase_details = {}
|
||
|
|
|
||
|
|
if purchase_upload_result.get("code") == 200:
|
||
|
|
purchase_task_id = purchase_upload_result.get("data", {}).get("taskId")
|
||
|
|
print(f" ✓ 采购交易上传成功, TaskID: {purchase_task_id}")
|
||
|
|
|
||
|
|
time.sleep(3)
|
||
|
|
|
||
|
|
# 查询导入状态
|
||
|
|
purchase_status_url = f"{BASE_URL}/ccdi/purchaseTransaction/importStatus/{purchase_task_id}"
|
||
|
|
purchase_status_result = client.get_import_status(purchase_status_url)
|
||
|
|
|
||
|
|
if purchase_status_result.get("code") == 200:
|
||
|
|
purchase_status_data = purchase_status_result.get("data", {})
|
||
|
|
|
||
|
|
# 查询失败记录
|
||
|
|
purchase_failures_url = f"{BASE_URL}/ccdi/purchaseTransaction/importFailures/{purchase_task_id}"
|
||
|
|
purchase_failures_result = client.get_import_failures(purchase_failures_url)
|
||
|
|
|
||
|
|
if purchase_failures_result.get("code") == 200:
|
||
|
|
purchase_failures = purchase_failures_result.get("rows", [])
|
||
|
|
|
||
|
|
purchase_details = {
|
||
|
|
"success_count": purchase_status_data.get("successCount", 0),
|
||
|
|
"failure_count": purchase_status_data.get("failureCount", 0),
|
||
|
|
"failures": purchase_failures
|
||
|
|
}
|
||
|
|
|
||
|
|
# 验证: 第1条失败(数据库重复), 第2条成功, 第3条失败(Excel内重复), 第4条成功
|
||
|
|
# 预期: 成功2条,失败2条
|
||
|
|
if purchase_status_data.get("successCount") == 2 and purchase_status_data.get("failureCount") == 2:
|
||
|
|
purchase_passed = True
|
||
|
|
print(f" ✓ 采购交易测试通过: 成功2条,失败2条")
|
||
|
|
else:
|
||
|
|
print(f" ✗ 采购交易测试失败: 预期成功2失败2, 实际成功{purchase_status_data.get('successCount')}失败{purchase_status_data.get('failureCount')}")
|
||
|
|
|
||
|
|
# 测试员工信息
|
||
|
|
print("\n >> 测试员工信息混合重复")
|
||
|
|
employee_upload_url = f"{BASE_URL}/ccdi/employee/importData"
|
||
|
|
employee_upload_result = client.upload_file(employee_upload_url, employee_file)
|
||
|
|
|
||
|
|
employee_passed = False
|
||
|
|
employee_details = {}
|
||
|
|
|
||
|
|
if employee_upload_result.get("code") == 200:
|
||
|
|
employee_task_id = employee_upload_result.get("data", {}).get("taskId")
|
||
|
|
print(f" ✓ 员工信息上传成功, TaskID: {employee_task_id}")
|
||
|
|
|
||
|
|
time.sleep(3)
|
||
|
|
|
||
|
|
# 查询导入状态
|
||
|
|
employee_status_url = f"{BASE_URL}/ccdi/employee/importStatus/{employee_task_id}"
|
||
|
|
employee_status_result = client.get_import_status(employee_status_url)
|
||
|
|
|
||
|
|
if employee_status_result.get("code") == 200:
|
||
|
|
employee_status_data = employee_status_result.get("data", {})
|
||
|
|
|
||
|
|
# 查询失败记录
|
||
|
|
employee_failures_url = f"{BASE_URL}/ccdi/employee/importFailures/{employee_task_id}"
|
||
|
|
employee_failures_result = client.get_import_failures(employee_failures_url)
|
||
|
|
|
||
|
|
if employee_failures_result.get("code") == 200:
|
||
|
|
employee_failures = employee_failures_result.get("rows", [])
|
||
|
|
|
||
|
|
employee_details = {
|
||
|
|
"success_count": employee_status_data.get("successCount", 0),
|
||
|
|
"failure_count": employee_status_data.get("failureCount", 0),
|
||
|
|
"failures": employee_failures
|
||
|
|
}
|
||
|
|
|
||
|
|
# 验证: 第1条失败(数据库重复), 第2条成功, 第3条失败(Excel内重复), 第4条成功
|
||
|
|
# 预期: 成功2条,失败2条
|
||
|
|
if employee_status_data.get("successCount") == 2 and employee_status_data.get("failureCount") == 2:
|
||
|
|
employee_passed = True
|
||
|
|
print(f" ✓ 员工信息测试通过: 成功2条,失败2条")
|
||
|
|
else:
|
||
|
|
print(f" ✗ 员工信息测试失败: 预期成功2失败2, 实际成功{employee_status_data.get('successCount')}失败{employee_status_data.get('failureCount')}")
|
||
|
|
|
||
|
|
self.details = {
|
||
|
|
"purchase": {
|
||
|
|
"passed": purchase_passed,
|
||
|
|
"details": purchase_details
|
||
|
|
},
|
||
|
|
"employee": {
|
||
|
|
"passed": employee_passed,
|
||
|
|
"details": employee_details
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
# 至少一个通过则认为测试通过(因为数据库可能不存在预置数据)
|
||
|
|
self.passed = purchase_passed or employee_passed
|
||
|
|
|
||
|
|
if self.passed:
|
||
|
|
print(f" ✓ 测试通过")
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
self.error_message = f"测试异常: {str(e)}"
|
||
|
|
print(f" ✗ 测试异常: {str(e)}")
|
||
|
|
|
||
|
|
self.end_time = datetime.now()
|
||
|
|
|
||
|
|
|
||
|
|
class TestRunner:
|
||
|
|
"""测试运行器"""
|
||
|
|
|
||
|
|
def __init__(self):
|
||
|
|
self.client = APIClient(BASE_URL)
|
||
|
|
self.test_cases: List[TestCase] = []
|
||
|
|
self.results = []
|
||
|
|
|
||
|
|
def add_test_case(self, test_case: TestCase):
|
||
|
|
"""添加测试用例"""
|
||
|
|
self.test_cases.append(test_case)
|
||
|
|
|
||
|
|
def run_all(self):
|
||
|
|
"""运行所有测试用例"""
|
||
|
|
print("=" * 80)
|
||
|
|
print("导入文件内部主键重复检测功能测试")
|
||
|
|
print("=" * 80)
|
||
|
|
print(f"测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||
|
|
print(f"测试环境: {BASE_URL}")
|
||
|
|
print("=" * 80)
|
||
|
|
|
||
|
|
# 登录
|
||
|
|
print("\n[1/2] 登录系统...")
|
||
|
|
if not self.client.login(USERNAME, PASSWORD):
|
||
|
|
print("✗ 登录失败,测试终止")
|
||
|
|
return
|
||
|
|
|
||
|
|
# 运行测试
|
||
|
|
print("\n[2/2] 运行测试用例...")
|
||
|
|
print("-" * 80)
|
||
|
|
|
||
|
|
for i, test_case in enumerate(self.test_cases, 1):
|
||
|
|
print(f"\n测试用例 {i}/{len(self.test_cases)}: {test_case.name}")
|
||
|
|
print(f"描述: {test_case.description}")
|
||
|
|
print("-" * 80)
|
||
|
|
|
||
|
|
test_case.run(self.client)
|
||
|
|
self.results.append(test_case.to_dict())
|
||
|
|
|
||
|
|
# 生成报告
|
||
|
|
self.generate_report()
|
||
|
|
|
||
|
|
def generate_report(self):
|
||
|
|
"""生成测试报告"""
|
||
|
|
print("\n" + "=" * 80)
|
||
|
|
print("测试报告")
|
||
|
|
print("=" * 80)
|
||
|
|
|
||
|
|
passed_count = sum(1 for r in self.results if r["passed"])
|
||
|
|
failed_count = len(self.results) - passed_count
|
||
|
|
|
||
|
|
print(f"\n总测试用例数: {len(self.results)}")
|
||
|
|
print(f"通过: {passed_count}")
|
||
|
|
print(f"失败: {failed_count}")
|
||
|
|
print(f"通过率: {passed_count / len(self.results) * 100:.1f}%")
|
||
|
|
|
||
|
|
print("\n详细结果:")
|
||
|
|
print("-" * 80)
|
||
|
|
|
||
|
|
for i, result in enumerate(self.results, 1):
|
||
|
|
status = "✓ PASS" if result["passed"] else "✗ FAIL"
|
||
|
|
print(f"\n{i}. {result['name']}")
|
||
|
|
print(f" 状态: {status}")
|
||
|
|
print(f" 耗时: {result['duration']}")
|
||
|
|
|
||
|
|
if not result["passed"]:
|
||
|
|
print(f" 错误: {result['error_message']}")
|
||
|
|
|
||
|
|
if result["details"]:
|
||
|
|
print(f" 详情: {json.dumps(result['details'], ensure_ascii=False, indent=6)}")
|
||
|
|
|
||
|
|
# 保存报告到文件
|
||
|
|
report_file = os.path.join(REPORT_DIR, f"test_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json")
|
||
|
|
with open(report_file, 'w', encoding='utf-8') as f:
|
||
|
|
json.dump({
|
||
|
|
"test_time": datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
|
||
|
|
"environment": BASE_URL,
|
||
|
|
"total_count": len(self.results),
|
||
|
|
"passed_count": passed_count,
|
||
|
|
"failed_count": failed_count,
|
||
|
|
"pass_rate": f"{passed_count / len(self.results) * 100:.1f}%",
|
||
|
|
"results": self.results
|
||
|
|
}, f, ensure_ascii=False, indent=2)
|
||
|
|
|
||
|
|
print(f"\n报告已保存到: {report_file}")
|
||
|
|
print("=" * 80)
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
"""主函数"""
|
||
|
|
runner = TestRunner()
|
||
|
|
|
||
|
|
# 添加测试用例
|
||
|
|
runner.add_test_case(PurchaseDuplicateTestCase())
|
||
|
|
runner.add_test_case(EmployeeEmployeeIdDuplicateTestCase())
|
||
|
|
runner.add_test_case(EmployeeIdCardDuplicateTestCase())
|
||
|
|
runner.add_test_case(MixedDuplicateTestCase())
|
||
|
|
|
||
|
|
# 运行所有测试
|
||
|
|
try:
|
||
|
|
runner.run_all()
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
print("\n\n测试被用户中断")
|
||
|
|
except Exception as e:
|
||
|
|
print(f"\n\n测试运行异常: {str(e)}")
|
||
|
|
import traceback
|
||
|
|
traceback.print_exc()
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
main()
|