修改目录
This commit is contained in:
308
assets/implementation/scripts/test_uniqueness_validation.py
Normal file
308
assets/implementation/scripts/test_uniqueness_validation.py
Normal file
@@ -0,0 +1,308 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
中介导入唯一性校验测试脚本
|
||||
|
||||
测试场景：
|
||||
1. 个人中介导入 - 证件号重复导入（非更新模式）应失败
|
||||
2. 个人中介导入 - 证件号重复导入（更新模式）应成功
|
||||
3. 机构中介导入 - 统一社会信用代码重复导入（非更新模式）应失败
|
||||
4. 机构中介导入 - 统一社会信用代码重复导入（更新模式）应成功
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
# 配置
|
||||
BASE_URL = "http://localhost:8080"
|
||||
USERNAME = "admin"
|
||||
PASSWORD = "admin123"
|
||||
|
||||
# 全局变量存储token
|
||||
token = None
|
||||
|
||||
|
||||
def login():
|
||||
"""登录获取token"""
|
||||
global token
|
||||
url = f"{BASE_URL}/login/test"
|
||||
data = {
|
||||
"username": USERNAME,
|
||||
"password": PASSWORD
|
||||
}
|
||||
response = requests.post(url, data=data)
|
||||
result = response.json()
|
||||
if result.get("code") == 200:
|
||||
token = result.get("token")
|
||||
print(f"✓ 登录成功，获取token: {token[:20]}...")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 登录失败: {result}")
|
||||
return False
|
||||
|
||||
|
||||
def get_headers():
|
||||
"""获取带token的请求头"""
|
||||
return {
|
||||
"Authorization": f"Bearer {token}"
|
||||
}
|
||||
|
||||
|
||||
def test_import_person_without_update(file_path, cert_no):
|
||||
"""
|
||||
测试场景1: 个人中介导入（非更新模式）- 证件号重复
|
||||
期望：导入失败，提示证件号已存在
|
||||
"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试场景1: 个人中介导入（非更新模式）- 证件号 {cert_no} 重复")
|
||||
print(f"{'='*60}")
|
||||
|
||||
url = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
||||
files = {"file": open(file_path, "rb")}
|
||||
data = {"updateSupport": "false"}
|
||||
|
||||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||||
result = response.json()
|
||||
|
||||
print(f"响应状态码: {response.status_code}")
|
||||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||||
|
||||
# 验证结果
|
||||
if result.get("code") == 500:
|
||||
if "已存在" in result.get("msg", ""):
|
||||
print(f"✓ 测试通过：系统正确拒绝了重复的证件号")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 测试失败：错误信息不符合预期")
|
||||
return False
|
||||
else:
|
||||
print(f"✗ 测试失败：系统应该拒绝重复的证件号")
|
||||
return False
|
||||
|
||||
|
||||
def test_import_person_with_update(file_path, cert_no):
|
||||
"""
|
||||
测试场景2: 个人中介导入（更新模式）- 证件号重复
|
||||
期望：导入成功，更新已存在的记录
|
||||
"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试场景2: 个人中介导入（更新模式）- 证件号 {cert_no} 重复")
|
||||
print(f"{'='*60}")
|
||||
|
||||
url = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
||||
files = {"file": open(file_path, "rb")}
|
||||
data = {"updateSupport": "true"}
|
||||
|
||||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||||
result = response.json()
|
||||
|
||||
print(f"响应状态码: {response.status_code}")
|
||||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||||
|
||||
# 验证结果
|
||||
if result.get("code") == 200:
|
||||
print(f"✓ 测试通过：系统成功更新了已存在的记录")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 测试失败：系统应该允许更新模式")
|
||||
return False
|
||||
|
||||
|
||||
def test_import_entity_without_update(file_path, credit_code):
|
||||
"""
|
||||
测试场景3: 机构中介导入（非更新模式）- 统一社会信用代码重复
|
||||
期望：导入失败，提示统一社会信用代码已存在
|
||||
"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试场景3: 机构中介导入（非更新模式）- 统一社会信用代码 {credit_code} 重复")
|
||||
print(f"{'='*60}")
|
||||
|
||||
url = f"{BASE_URL}/dpc/intermediary/importEntityData"
|
||||
files = {"file": open(file_path, "rb")}
|
||||
data = {"updateSupport": "false"}
|
||||
|
||||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||||
result = response.json()
|
||||
|
||||
print(f"响应状态码: {response.status_code}")
|
||||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||||
|
||||
# 验证结果
|
||||
if result.get("code") == 500:
|
||||
if "已存在" in result.get("msg", ""):
|
||||
print(f"✓ 测试通过：系统正确拒绝了重复的统一社会信用代码")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 测试失败：错误信息不符合预期")
|
||||
return False
|
||||
else:
|
||||
print(f"✗ 测试失败：系统应该拒绝重复的统一社会信用代码")
|
||||
return False
|
||||
|
||||
|
||||
def test_import_entity_with_update(file_path, credit_code):
|
||||
"""
|
||||
测试场景4: 机构中介导入（更新模式）- 统一社会信用代码重复
|
||||
期望：导入成功，更新已存在的记录
|
||||
"""
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试场景4: 机构中介导入（更新模式）- 统一社会信用代码 {credit_code} 重复")
|
||||
print(f"{'='*60}")
|
||||
|
||||
url = f"{BASE_URL}/dpc/intermediary/importEntityData"
|
||||
files = {"file": open(file_path, "rb")}
|
||||
data = {"updateSupport": "true"}
|
||||
|
||||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||||
result = response.json()
|
||||
|
||||
print(f"响应状态码: {response.status_code}")
|
||||
print(f"响应内容: {json.dumps(result, ensure_ascii=False, indent=2)}")
|
||||
|
||||
# 验证结果
|
||||
if result.get("code") == 200:
|
||||
print(f"✓ 测试通过：系统成功更新了已存在的记录")
|
||||
return True
|
||||
else:
|
||||
print(f"✗ 测试失败：系统应该允许更新模式")
|
||||
return False
|
||||
|
||||
|
||||
def create_test_person_excel(file_path, cert_no, name="测试用户"):
|
||||
"""创建测试用的个人中介Excel文件"""
|
||||
import openpyxl
|
||||
from openpyxl.styles import Protection
|
||||
|
||||
wb = openpyxl.Workbook()
|
||||
ws = wb.active
|
||||
ws.title = "个人中介黑名单"
|
||||
|
||||
# 表头
|
||||
headers = ["姓名", "证件号码", "人员类型", "人员子类型", "性别", "证件类型", "手机号", "微信号",
|
||||
"联系地址", "所在公司", "职位", "关联人员ID", "关联关系", "备注"]
|
||||
ws.append(headers)
|
||||
|
||||
# 添加测试数据
|
||||
ws.append([name, cert_no, "中介", "本人", "男", "身份证", "13800138000",
|
||||
"test_wxh", "测试地址", "测试公司", "测试职位", "", "", "测试备注"])
|
||||
|
||||
wb.save(file_path)
|
||||
print(f"✓ 创建测试Excel文件: {file_path}")
|
||||
|
||||
|
||||
def create_test_entity_excel(file_path, credit_code, name="测试机构"):
|
||||
"""创建测试用的机构中介Excel文件"""
|
||||
import openpyxl
|
||||
|
||||
wb = openpyxl.Workbook()
|
||||
ws = wb.active
|
||||
ws.title = "机构中介黑名单"
|
||||
|
||||
# 表头
|
||||
headers = ["机构名称", "统一社会信用代码", "主体类型", "企业性质", "行业分类", "所属行业", "成立日期",
|
||||
"注册地址", "法定代表人", "法定代表人证件类型", "法定代表人证件号码", "股东1", "股东2",
|
||||
"股东3", "股东4", "股东5", "备注"]
|
||||
ws.append(headers)
|
||||
|
||||
# 添加测试数据
|
||||
ws.append([name, credit_code, "有限责任公司", "民企", "金融业", "银行业", "2020-01-01",
|
||||
"北京市测试区测试路123号", "张三", "身份证", "110101199001011234",
|
||||
"股东A", "股东B", "股东C", "股东D", "股东E", "测试备注"])
|
||||
|
||||
wb.save(file_path)
|
||||
print(f"✓ 创建测试Excel文件: {file_path}")
|
||||
|
||||
|
||||
def main():
|
||||
"""主测试流程"""
|
||||
print(f"\n{'#'*60}")
|
||||
print(f"# 中介导入唯一性校验测试")
|
||||
print(f"# 测试时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
print(f"{'#'*60}")
|
||||
|
||||
# 登录
|
||||
if not login():
|
||||
return
|
||||
|
||||
# 测试参数
|
||||
test_cert_no = f"TEST{int(time.time())}" # 生成唯一测试证件号
|
||||
test_credit_code = f"91{int(time.time())}001" # 生成唯一测试统一社会信用代码
|
||||
|
||||
test_results = []
|
||||
|
||||
# 准备测试文件
|
||||
person_file = "test_person_uniqueness.xlsx"
|
||||
entity_file = "test_entity_uniqueness.xlsx"
|
||||
|
||||
# ========== 场景1: 先导入一条个人数据 ==========
|
||||
print(f"\n{'='*60}")
|
||||
print(f"准备步骤: 首次导入个人中介数据（证件号: {test_cert_no}）")
|
||||
print(f"{'='*60}")
|
||||
|
||||
create_test_person_excel(person_file, test_cert_no)
|
||||
|
||||
url = f"{BASE_URL}/dpc/intermediary/importPersonData"
|
||||
files = {"file": open(person_file, "rb")}
|
||||
data = {"updateSupport": "false"}
|
||||
|
||||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||||
result = response.json()
|
||||
|
||||
if result.get("code") == 200:
|
||||
print(f"✓ 首次导入成功")
|
||||
else:
|
||||
print(f"✗ 首次导入失败: {result}")
|
||||
return
|
||||
|
||||
# ========== 场景1: 非更新模式导入重复个人数据 ==========
|
||||
test_results.append(test_import_person_without_update(person_file, test_cert_no))
|
||||
|
||||
# ========== 场景2: 更新模式导入重复个人数据 ==========
|
||||
test_results.append(test_import_person_with_update(person_file, test_cert_no))
|
||||
|
||||
# ========== 准备: 首次导入机构数据 ==========
|
||||
print(f"\n{'='*60}")
|
||||
print(f"准备步骤: 首次导入机构中介数据（统一社会信用代码: {test_credit_code}）")
|
||||
print(f"{'='*60}")
|
||||
|
||||
create_test_entity_excel(entity_file, test_credit_code)
|
||||
|
||||
url = f"{BASE_URL}/dpc/intermediary/importEntityData"
|
||||
files = {"file": open(entity_file, "rb")}
|
||||
data = {"updateSupport": "false"}
|
||||
|
||||
response = requests.post(url, files=files, data=data, headers=get_headers())
|
||||
result = response.json()
|
||||
|
||||
if result.get("code") == 200:
|
||||
print(f"✓ 首次导入成功")
|
||||
else:
|
||||
print(f"✗ 首次导入失败: {result}")
|
||||
return
|
||||
|
||||
# ========== 场景3: 非更新模式导入重复机构数据 ==========
|
||||
test_results.append(test_import_entity_without_update(entity_file, test_credit_code))
|
||||
|
||||
# ========== 场景4: 更新模式导入重复机构数据 ==========
|
||||
test_results.append(test_import_entity_with_update(entity_file, test_credit_code))
|
||||
|
||||
# ========== 输出测试报告 ==========
|
||||
print(f"\n{'='*60}")
|
||||
print(f"测试报告汇总")
|
||||
print(f"{'='*60}")
|
||||
print(f"测试场景总数: {len(test_results)}")
|
||||
print(f"通过数量: {sum(test_results)}")
|
||||
print(f"失败数量: {len(test_results) - sum(test_results)}")
|
||||
|
||||
if all(test_results):
|
||||
print(f"\n✓ 所有测试通过！")
|
||||
else:
|
||||
print(f"\n✗ 部分测试失败，请查看上方详细日志")
|
||||
|
||||
print(f"\n测试完成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user